Vito
ab333d3838
* feat(tags): keep unused tags in the shared vocabulary * feat(vocabulary): add bounded tag and category reads * feat(vocabulary): guard tag create, rename and merge * feat(vocabulary): preview and apply category consolidation * refactor(vocabulary): share canonical token hashing * test(vocabulary): prove retained tags survive restore * test(vocabulary): cover category paging tie-break and clarify contract docs * test(vocabulary): assert rejected tag creations write nothing
829 lines
33 KiB
Python
829 lines
33 KiB
Python
"""Bounded vocabulary reads count exactly what Library Browse shows."""
|
|
|
|
import re
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from database.library_commands import LibraryCommandError
|
|
from database.operations import PromptDatabase
|
|
from utils.hashing import generate_prompt_hash
|
|
|
|
|
|
class TestVocabularyReads(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.db = PromptDatabase(str(Path(self.temp.name) / "vocabulary.db"))
|
|
|
|
def tearDown(self):
|
|
try:
|
|
# Windows refuses to remove a database that still has open handles.
|
|
self.db.close_all()
|
|
finally:
|
|
self.temp.cleanup()
|
|
|
|
def save(self, text, **metadata):
|
|
return self.db.save_prompt(
|
|
text, prompt_hash=generate_prompt_hash(text), **metadata
|
|
)
|
|
|
|
def read(self, name, **options):
|
|
method = getattr(self.db, name, None)
|
|
self.assertTrue(callable(method), f"PromptDatabase.{name} is missing")
|
|
return method(**options)
|
|
|
|
def add_unused(self, *names):
|
|
with self.db.model.get_connection() as conn:
|
|
for name in names:
|
|
conn.execute("INSERT INTO tags(name) VALUES(?)", (name,))
|
|
|
|
def all_tags(self):
|
|
rows, after = [], None
|
|
while True:
|
|
page = self.read("list_vocabulary_tags", limit=200, after=after)
|
|
rows.extend(page["tags"])
|
|
if not page["has_more"]:
|
|
return rows
|
|
after = page["next_key"]
|
|
|
|
def test_tag_rows_are_registry_rows_in_creation_order_with_unused_at_zero(self):
|
|
self.save("First", tags=["c"])
|
|
self.save("Second", tags=["a"])
|
|
self.save("Third", tags=["b", "a"])
|
|
self.add_unused("z")
|
|
page = self.read("list_vocabulary_tags")
|
|
self.assertEqual([row["name"] for row in page["tags"]], ["c", "a", "b", "z"])
|
|
self.assertEqual(page["total"], 4)
|
|
self.assertFalse(page["has_more"])
|
|
self.assertIsNone(page["next_key"])
|
|
by_name = {row["name"]: row for row in page["tags"]}
|
|
self.assertEqual(by_name["z"]["prompt_count"], 0)
|
|
self.assertEqual(by_name["a"]["prompt_count"], 2)
|
|
self.assertEqual(set(by_name["c"]), {"id", "name", "prompt_count"})
|
|
|
|
def test_prompt_count_equals_library_browse_total_for_every_tag(self):
|
|
self.save("Parent only", tags=["parent-only"])
|
|
self.db.save_prompt_pair("Pair cat", "Pair noise", tags=["pair-only"])
|
|
archived = self.db.save_prompt_pair(
|
|
"Archived cat", "Archived noise", tags=["archived-only"]
|
|
)
|
|
self.db.archive_prompt_pair(archived)
|
|
self.save("Legacy blur", tags=["negative-only"])
|
|
self.db.save_prompt_pair("Negative owner", "Legacy blur")
|
|
self.save("Dual dog", tags=["dual"])
|
|
self.db.save_prompt_pair("Dual dog", "Dual noise", tags=["dual"])
|
|
retained = self.save("Retained negative", tags=["retained"])
|
|
self.db.save_prompt_pair("Retained owner", "Retained negative")
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
|
|
(retained, "Kept"),
|
|
)
|
|
self.add_unused("unused")
|
|
rows = self.all_tags()
|
|
self.assertEqual(len(rows), 7)
|
|
for row in rows:
|
|
self.assertEqual(
|
|
row["prompt_count"],
|
|
self.db.list_library(tags=[row["name"]])["total"],
|
|
row["name"],
|
|
)
|
|
counts = {row["name"]: row["prompt_count"] for row in rows}
|
|
self.assertEqual(
|
|
counts,
|
|
{
|
|
"parent-only": 1,
|
|
"pair-only": 1,
|
|
"archived-only": 0,
|
|
"negative-only": 0,
|
|
"dual": 1,
|
|
"retained": 1,
|
|
"unused": 0,
|
|
},
|
|
)
|
|
|
|
def test_category_rows_match_library_options_and_browse_totals(self):
|
|
self.save("Zeta first", category="Zeta")
|
|
self.db.save_prompt_pair("Alpha owner", "Alpha noise", category="Alpha")
|
|
self.save("Padded", category=" Landscape ")
|
|
self.save("Plain", category="Landscape")
|
|
self.save("Blank", category=" ")
|
|
self.save("Null")
|
|
self.save("Zeta again", category="Zeta")
|
|
archived = self.db.save_prompt_pair(
|
|
"Archived owner", "Archived noise", category="Archived-only"
|
|
)
|
|
self.db.archive_prompt_pair(archived)
|
|
self.save("Legacy negative", category="Negative-only")
|
|
self.db.save_prompt_pair("Negative owner", "Legacy negative")
|
|
page = self.read("list_vocabulary_categories")
|
|
values = [row["value"] for row in page["categories"]]
|
|
self.assertEqual(values, ["Zeta", "Alpha", " Landscape ", "Landscape"])
|
|
self.assertEqual(set(values), set(self.db.get_library_options()["categories"]))
|
|
self.assertEqual(page["total"], 4)
|
|
self.assertFalse(page["has_more"])
|
|
for row in page["categories"]:
|
|
self.assertEqual(set(row), {"value", "prompt_count"})
|
|
self.assertEqual(
|
|
row["prompt_count"],
|
|
self.db.list_library(category=row["value"])["total"],
|
|
row["value"],
|
|
)
|
|
self.assertEqual(page["categories"][0]["prompt_count"], 2)
|
|
|
|
def test_category_pages_continue_by_first_use_then_value(self):
|
|
for index in range(5):
|
|
self.save(f"Owner {index}", category=f"C{4 - index}")
|
|
first = self.read("list_vocabulary_categories", limit=2)
|
|
second = self.read(
|
|
"list_vocabulary_categories", limit=2, after=first["next_key"]
|
|
)
|
|
third = self.read(
|
|
"list_vocabulary_categories", limit=2, after=second["next_key"]
|
|
)
|
|
values = [
|
|
row["value"]
|
|
for page in (first, second, third)
|
|
for row in page["categories"]
|
|
]
|
|
self.assertEqual(values, ["C4", "C3", "C2", "C1", "C0"])
|
|
self.assertEqual(
|
|
[page["has_more"] for page in (first, second, third)], [True, True, False]
|
|
)
|
|
for after in ([1], ["1", "C4"], {"id": 1}, [True, "C4"]):
|
|
with self.assertRaises(ValueError):
|
|
self.read("list_vocabulary_categories", after=after)
|
|
|
|
def test_category_paging_breaks_first_use_ties_by_value(self):
|
|
self.save("Shared owner", category="B")
|
|
self.db.save_prompt_pair(
|
|
"Shared owner", "Pair noise A", category="A", title="A pair"
|
|
)
|
|
self.db.save_prompt_pair(
|
|
"Shared owner", "Pair noise C", category="C", title="C pair"
|
|
)
|
|
pages, after = [], None
|
|
for _ in range(3):
|
|
page = self.read("list_vocabulary_categories", limit=1, after=after)
|
|
pages.append(page)
|
|
after = page["next_key"]
|
|
values = [row["value"] for page in pages for row in page["categories"]]
|
|
self.assertEqual(values, ["A", "B", "C"])
|
|
self.assertEqual([page["has_more"] for page in pages], [True, True, False])
|
|
for page in pages:
|
|
row = page["categories"][0]
|
|
self.assertEqual(
|
|
row["prompt_count"],
|
|
self.db.list_library(category=row["value"])["total"],
|
|
)
|
|
|
|
def test_pages_are_bounded_and_continue_without_gaps(self):
|
|
self.add_unused(*[f"tag-{index:03d}" for index in range(250)])
|
|
pages, after = [], None
|
|
for _ in range(3):
|
|
page = self.read("list_vocabulary_tags", limit=100, after=after)
|
|
pages.append(page)
|
|
after = page["next_key"]
|
|
self.assertEqual([len(page["tags"]) for page in pages], [100, 100, 50])
|
|
self.assertEqual([page["has_more"] for page in pages], [True, True, False])
|
|
self.assertEqual({page["total"] for page in pages}, {250})
|
|
names = [row["name"] for page in pages for row in page["tags"]]
|
|
self.assertEqual(names, [f"tag-{index:03d}" for index in range(250)])
|
|
for options in ({"limit": 0}, {"limit": 201}, {"after": "1"}, {"after": 0}):
|
|
with self.assertRaises(ValueError):
|
|
self.read("list_vocabulary_tags", **options)
|
|
|
|
def test_reads_share_one_snapshot_and_report_epoch(self):
|
|
self.save("Cat", category="Animal", tags=["cat"])
|
|
epoch = self.db.get_execution_epoch()
|
|
tags = self.read("list_vocabulary_tags")
|
|
categories = self.read("list_vocabulary_categories")
|
|
self.assertEqual(tags["library_epoch"], epoch)
|
|
self.assertEqual(categories["library_epoch"], epoch)
|
|
helper = getattr(self.db, "_vocabulary_tags", None)
|
|
self.assertTrue(callable(helper), "PromptDatabase._vocabulary_tags is missing")
|
|
with self.db.library_snapshot() as (conn, snapshot_epoch, _ceiling):
|
|
inner = helper(conn, snapshot_epoch, limit=10, after=None)
|
|
self.assertEqual(inner["library_epoch"], epoch)
|
|
self.assertTrue(conn.in_transaction)
|
|
|
|
|
|
_TAG_TABLES = ('"tags"', '"prompt_tags"', '"prompt_pair_tags"', '"sqlite_sequence"')
|
|
|
|
|
|
class TestVocabularyTagCommands(unittest.TestCase):
|
|
"""Guarded tag create, rename, merge and delete across both junctions."""
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.db = PromptDatabase(str(Path(self.temp.name) / "commands.db"))
|
|
self.epoch = self.db.get_execution_epoch()
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.db.close_all()
|
|
finally:
|
|
self.temp.cleanup()
|
|
|
|
def command(self, method_name, /, *args, **options):
|
|
method = getattr(self.db, method_name, None)
|
|
self.assertTrue(callable(method), f"PromptDatabase.{method_name} is missing")
|
|
return method(*args, **options)
|
|
|
|
def save(self, text, **metadata):
|
|
return self.db.save_prompt(
|
|
text, prompt_hash=generate_prompt_hash(text), **metadata
|
|
)
|
|
|
|
def tag_id(self, name):
|
|
with self.db.model.get_connection() as conn:
|
|
row = conn.execute("SELECT id FROM tags WHERE name=?", (name,)).fetchone()
|
|
return row[0] if row else None
|
|
|
|
def links(self, tag_name):
|
|
with self.db.model.get_connection() as conn:
|
|
components = conn.execute(
|
|
"SELECT pt.prompt_id FROM prompt_tags pt JOIN tags t "
|
|
"ON t.id=pt.tag_id WHERE t.name=? ORDER BY 1",
|
|
(tag_name,),
|
|
).fetchall()
|
|
pairs = conn.execute(
|
|
"SELECT ppt.pair_id FROM prompt_pair_tags ppt JOIN tags t "
|
|
"ON t.id=ppt.tag_id WHERE t.name=? ORDER BY 1",
|
|
(tag_name,),
|
|
).fetchall()
|
|
return [row[0] for row in components], [row[0] for row in pairs]
|
|
|
|
def junction_rows(self):
|
|
with self.db.model.get_connection() as conn:
|
|
return (
|
|
conn.execute("SELECT * FROM tags ORDER BY id").fetchall(),
|
|
conn.execute("SELECT * FROM prompt_tags ORDER BY 1,2").fetchall(),
|
|
conn.execute("SELECT * FROM prompt_pair_tags ORDER BY 1,2").fetchall(),
|
|
)
|
|
|
|
def other_rows(self):
|
|
"""Every row outside the tag registry and junctions, timestamps included."""
|
|
conn = self.db.model.get_connection()
|
|
return [
|
|
line
|
|
for line in conn.iterdump()
|
|
if line.startswith("INSERT INTO")
|
|
and not line.startswith(tuple(f"INSERT INTO {t}" for t in _TAG_TABLES))
|
|
]
|
|
|
|
def changes(self):
|
|
return self.db.model.get_connection().total_changes
|
|
|
|
def owners(self, tag="moody"):
|
|
"""Parent, active pair, archived pair and negative-only component."""
|
|
parent = self.save("Parent sky", category="Scenery", tags=[tag])
|
|
self.db.link_image_to_prompt(parent, str(Path(self.temp.name) / "a.png"))
|
|
active = self.db.save_prompt_pair("Active sky", "Active noise", tags=[tag])
|
|
archived = self.db.save_prompt_pair(
|
|
"Archived sky", "Archived noise", tags=[tag]
|
|
)
|
|
self.db.archive_prompt_pair(archived)
|
|
negative = self.save("Legacy blur", tags=[tag])
|
|
self.db.save_prompt_pair("Negative owner", "Legacy blur")
|
|
return parent, negative, active, archived
|
|
|
|
def test_create_inserts_unused_tag_and_reuses_exact_existing(self):
|
|
self.save("Owner", tags=["other"])
|
|
before = self.junction_rows()
|
|
changes = self.changes()
|
|
created = self.command(
|
|
"create_vocabulary_tag", " sunset ", expected_epoch=self.epoch
|
|
)
|
|
self.assertEqual(self.changes() - changes, 1)
|
|
self.assertEqual(created["library_epoch"], self.epoch)
|
|
self.assertTrue(created["created"])
|
|
self.assertEqual(
|
|
created["tag"],
|
|
{"id": self.tag_id("sunset"), "name": "sunset", "prompt_count": 0},
|
|
)
|
|
after = self.junction_rows()
|
|
self.assertEqual(after[1:], before[1:])
|
|
changes = self.changes()
|
|
again = self.command(
|
|
"create_vocabulary_tag", "sunset", expected_epoch=self.epoch
|
|
)
|
|
self.assertEqual(self.changes(), changes)
|
|
self.assertFalse(again["created"])
|
|
self.assertEqual(again["tag"], created["tag"])
|
|
other = self.command(
|
|
"create_vocabulary_tag", "Sunset", expected_epoch=self.epoch
|
|
)
|
|
self.assertTrue(other["created"])
|
|
self.assertNotEqual(other["tag"]["id"], created["tag"]["id"])
|
|
self.assertIn("sunset", self.db.get_all_tags())
|
|
|
|
def test_create_validates_names(self):
|
|
before = self.junction_rows()
|
|
for name in ("", " ", "x" * 51, "bad\x01", "a,b", 7, None, ["a"]):
|
|
with self.assertRaises(ValueError, msg=repr(name)):
|
|
self.command("create_vocabulary_tag", name, expected_epoch=self.epoch)
|
|
with self.assertRaises(ValueError):
|
|
self.command("create_vocabulary_tag", "ok", expected_epoch="nope")
|
|
self.assertEqual(self.junction_rows(), before)
|
|
|
|
def test_rename_to_new_name_keeps_id_and_all_links(self):
|
|
parent, negative, active, archived = self.owners()
|
|
tag_id = self.tag_id("moody")
|
|
others = self.other_rows()
|
|
result = self.command(
|
|
"rename_vocabulary_tag",
|
|
tag_id,
|
|
expected_epoch=self.epoch,
|
|
name="moody",
|
|
new_name=" dramatic ",
|
|
)
|
|
self.assertEqual(
|
|
result,
|
|
{
|
|
"library_epoch": self.epoch,
|
|
"merged": False,
|
|
"tag": {"id": tag_id, "name": "dramatic", "prompt_count": 2},
|
|
"components_changed": 2,
|
|
"pairs_changed": 2,
|
|
},
|
|
)
|
|
self.assertIsNone(self.tag_id("moody"))
|
|
self.assertEqual(
|
|
self.links("dramatic"), ([parent, negative], [active, archived])
|
|
)
|
|
self.assertEqual(self.db.get_prompt_pair(archived)["tags"], ["dramatic"])
|
|
self.assertEqual(self.other_rows(), others)
|
|
|
|
def test_rename_to_existing_merges_both_junctions_without_duplicates(self):
|
|
both = self.save("Both", tags=["old", "new"])
|
|
only_old = self.save("Old only", tags=["old"])
|
|
pair_both = self.db.save_prompt_pair("Pair both", "Noise", tags=["old", "new"])
|
|
pair_old = self.db.save_prompt_pair("Pair old", "Noise", tags=["old"])
|
|
self.db.archive_prompt_pair(pair_old)
|
|
source, target = self.tag_id("old"), self.tag_id("new")
|
|
others = self.other_rows()
|
|
result = self.command(
|
|
"rename_vocabulary_tag",
|
|
source,
|
|
expected_epoch=self.epoch,
|
|
name="old",
|
|
new_name="new",
|
|
)
|
|
self.assertTrue(result["merged"])
|
|
self.assertEqual(result["tag"]["id"], target)
|
|
self.assertEqual(result["tag"]["name"], "new")
|
|
self.assertEqual(result["tag"]["prompt_count"], 3)
|
|
self.assertEqual(
|
|
(result["components_changed"], result["pairs_changed"]), (2, 2)
|
|
)
|
|
self.assertIsNone(self.tag_id("old"))
|
|
self.assertEqual(self.links("new"), ([both, only_old], [pair_both, pair_old]))
|
|
self.assertEqual(self.db.get_prompt_by_id(both)["tags"], ["new"])
|
|
self.assertEqual(self.db.get_prompt_pair(pair_both)["tags"], ["new"])
|
|
self.assertEqual(self.other_rows(), others)
|
|
|
|
def test_case_only_rename_is_a_rename(self):
|
|
prompt = self.save("Kitten", tags=["Cat"])
|
|
tag_id = self.tag_id("Cat")
|
|
result = self.command(
|
|
"rename_vocabulary_tag",
|
|
tag_id,
|
|
expected_epoch=self.epoch,
|
|
name="Cat",
|
|
new_name="cat",
|
|
)
|
|
self.assertFalse(result["merged"])
|
|
self.assertEqual(
|
|
result["tag"], {"id": tag_id, "name": "cat", "prompt_count": 1}
|
|
)
|
|
self.assertEqual(self.db.get_prompt_by_id(prompt)["tags"], ["cat"])
|
|
|
|
def test_stale_identity_and_epoch_write_nothing(self):
|
|
self.owners()
|
|
tag_id = self.tag_id("moody")
|
|
before = (self.junction_rows(), self.other_rows())
|
|
changes = self.changes()
|
|
for tag, name in ((tag_id, "Moody"), (tag_id + 99, "moody")):
|
|
for method, extra in (
|
|
("rename_vocabulary_tag", {"new_name": "fresh"}),
|
|
("delete_vocabulary_tag", {}),
|
|
):
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.command(
|
|
method, tag, expected_epoch=self.epoch, name=name, **extra
|
|
)
|
|
self.assertEqual(caught.exception.code, "vocabulary-conflict")
|
|
self.assertEqual(caught.exception.epoch, self.epoch)
|
|
stale = "00000000-0000-4000-8000-000000000000"
|
|
with patch.object(type(self.db), "_vocabulary_tag", create=True) as lookup:
|
|
for method, args, extra in (
|
|
("create_vocabulary_tag", ("fresh",), {}),
|
|
(
|
|
"rename_vocabulary_tag",
|
|
(tag_id,),
|
|
{"name": "moody", "new_name": "x"},
|
|
),
|
|
("delete_vocabulary_tag", (tag_id,), {"name": "moody"}),
|
|
):
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.command(method, *args, expected_epoch=stale, **extra)
|
|
self.assertEqual(caught.exception.code, "storage-changed")
|
|
self.assertEqual(caught.exception.epoch, self.epoch)
|
|
lookup.assert_not_called()
|
|
for new_name in ("moody", " moody "):
|
|
with self.assertRaises(ValueError):
|
|
self.command(
|
|
"rename_vocabulary_tag",
|
|
tag_id,
|
|
expected_epoch=self.epoch,
|
|
name="moody",
|
|
new_name=new_name,
|
|
)
|
|
for bad_id in (0, -1, "1", True, 2**63):
|
|
with self.assertRaises(ValueError):
|
|
self.command(
|
|
"delete_vocabulary_tag",
|
|
bad_id,
|
|
expected_epoch=self.epoch,
|
|
name="moody",
|
|
)
|
|
self.assertEqual(self.changes(), changes)
|
|
self.assertEqual((self.junction_rows(), self.other_rows()), before)
|
|
|
|
def test_delete_removes_name_and_links_only(self):
|
|
self.owners()
|
|
self.save("Bystander", tags=["keep"])
|
|
tag_id = self.tag_id("moody")
|
|
impact = self.command("vocabulary_tag_impact", tag_id)
|
|
self.assertEqual(
|
|
impact,
|
|
{
|
|
"library_epoch": self.epoch,
|
|
"impact": {
|
|
"positive_prompts": 1,
|
|
"other_prompts": 1,
|
|
"negative_variants": 1,
|
|
"archived_negative_variants": 1,
|
|
},
|
|
},
|
|
)
|
|
self.assertIsNone(self.command("vocabulary_tag_impact", tag_id + 99))
|
|
others = self.other_rows()
|
|
result = self.command(
|
|
"delete_vocabulary_tag", tag_id, expected_epoch=self.epoch, name="moody"
|
|
)
|
|
self.assertEqual(
|
|
result,
|
|
{
|
|
"library_epoch": self.epoch,
|
|
"deleted": True,
|
|
"components_changed": 2,
|
|
"pairs_changed": 2,
|
|
},
|
|
)
|
|
self.assertIsNone(self.tag_id("moody"))
|
|
self.assertEqual(self.links("moody"), ([], []))
|
|
self.assertEqual(self.other_rows(), others)
|
|
self.assertEqual(len(self.links("keep")[0]), 1)
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.command(
|
|
"delete_vocabulary_tag", tag_id, expected_epoch=self.epoch, name="moody"
|
|
)
|
|
self.assertEqual(caught.exception.code, "vocabulary-conflict")
|
|
|
|
|
|
_OLD_STAMP = "2000-01-01T00:00:00+00:00"
|
|
_DML = re.compile(r"\s*(INSERT|UPDATE|DELETE|REPLACE)\b", re.IGNORECASE)
|
|
|
|
|
|
class TestVocabularyCategoryCommands(unittest.TestCase):
|
|
"""Category preview and token-guarded rename/consolidation."""
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.db = PromptDatabase(str(Path(self.temp.name) / "categories.db"))
|
|
self.epoch = self.db.get_execution_epoch()
|
|
|
|
def tearDown(self):
|
|
try:
|
|
self.db.close_all()
|
|
finally:
|
|
self.temp.cleanup()
|
|
|
|
def command(self, method_name, /, **options):
|
|
method = getattr(self.db, method_name, None)
|
|
self.assertTrue(callable(method), f"PromptDatabase.{method_name} is missing")
|
|
return method(**options)
|
|
|
|
def preview(self, source="Scenery", target="Landscape"):
|
|
return self.command(
|
|
"preview_vocabulary_category",
|
|
expected_epoch=self.epoch,
|
|
source=source,
|
|
target=target,
|
|
)
|
|
|
|
def apply(self, token, source="Scenery", target="Landscape"):
|
|
return self.command(
|
|
"apply_vocabulary_category",
|
|
expected_epoch=self.epoch,
|
|
source=source,
|
|
target=target,
|
|
impact_token=token,
|
|
)
|
|
|
|
def save(self, text, **metadata):
|
|
return self.db.save_prompt(
|
|
text, prompt_hash=generate_prompt_hash(text), **metadata
|
|
)
|
|
|
|
def owners(self, source="Scenery"):
|
|
"""2 eligible, 1 negative-only component; 2 active, 1 archived pair."""
|
|
first = self.save("Parent one", category=source)
|
|
second = self.save("Parent two", category=source)
|
|
negative = self.save("Legacy blur", category=source)
|
|
self.db.save_prompt_pair("Negative owner", "Legacy blur")
|
|
active = [
|
|
self.db.save_prompt_pair(
|
|
f"Active {n}",
|
|
f"Noise {n}",
|
|
category=source,
|
|
title=f"Title {n}",
|
|
tags=[f"tag-{n}"],
|
|
rating=n,
|
|
notes=f"Notes {n}",
|
|
)
|
|
for n in (1, 2)
|
|
]
|
|
archived = self.db.save_prompt_pair(
|
|
"Archived sky", "Archived noise", category=source, title="Kept"
|
|
)
|
|
self.db.archive_prompt_pair(archived)
|
|
return [first, second], negative, active, archived
|
|
|
|
def age_rows(self):
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute("UPDATE prompts SET updated_at=?", (_OLD_STAMP,))
|
|
conn.execute("UPDATE prompt_pairs SET updated_at=?", (_OLD_STAMP,))
|
|
|
|
def rows(self, table):
|
|
with self.db.model.get_connection() as conn:
|
|
return {
|
|
row["id"]: dict(row)
|
|
for row in conn.execute(f"SELECT * FROM {table} ORDER BY id")
|
|
}
|
|
|
|
def snapshot(self):
|
|
return self.rows("prompts"), self.rows("prompt_pairs")
|
|
|
|
def traced(self, call):
|
|
statements = []
|
|
original = self.db.model.get_connection
|
|
|
|
def trace_connection():
|
|
connection = original()
|
|
connection.set_trace_callback(statements.append)
|
|
return connection
|
|
|
|
with patch.object(
|
|
self.db.model, "get_connection", side_effect=trace_connection
|
|
):
|
|
result = call()
|
|
conn = original()
|
|
conn.set_trace_callback(None)
|
|
return result, statements
|
|
|
|
def test_preview_counts_every_owner_type(self):
|
|
self.owners()
|
|
self.save("Elsewhere", category="Other")
|
|
before = self.snapshot()
|
|
result, statements = self.traced(self.preview)
|
|
self.assertEqual(
|
|
set(result),
|
|
{
|
|
"library_epoch",
|
|
"source",
|
|
"target",
|
|
"target_exists",
|
|
"impact",
|
|
"impact_token",
|
|
},
|
|
)
|
|
self.assertEqual(result["library_epoch"], self.epoch)
|
|
self.assertEqual((result["source"], result["target"]), ("Scenery", "Landscape"))
|
|
self.assertEqual(
|
|
result["impact"],
|
|
{
|
|
"positive_prompts": 2,
|
|
"other_prompts": 1,
|
|
"negative_variants": 2,
|
|
"archived_negative_variants": 1,
|
|
},
|
|
)
|
|
self.assertFalse(result["target_exists"])
|
|
self.assertRegex(result["impact_token"], r"\A[0-9a-f]{64}\Z")
|
|
self.assertEqual([q for q in statements if _DML.match(q)], [])
|
|
self.assertEqual(self.snapshot(), before)
|
|
trimmed = self.preview(target=" Other ")
|
|
self.assertEqual(trimmed["target"], "Other")
|
|
self.assertTrue(trimmed["target_exists"])
|
|
self.assertNotEqual(trimmed["impact_token"], result["impact_token"])
|
|
self.assertEqual(self.preview(), result)
|
|
|
|
def test_apply_renames_exact_value_on_components_and_all_pairs(self):
|
|
components, negative, active, archived = self.owners()
|
|
variant = self.save("Padded", category=" Scenery ")
|
|
bystander = self.save("Bystander", category="Other")
|
|
self.age_rows()
|
|
prompts_before, pairs_before = self.snapshot()
|
|
preview = self.preview()
|
|
receipt = self.apply(preview["impact_token"])
|
|
self.assertEqual(
|
|
receipt,
|
|
{
|
|
"library_epoch": self.epoch,
|
|
"source": "Scenery",
|
|
"target": "Landscape",
|
|
"consolidated": False,
|
|
"impact": preview["impact"],
|
|
},
|
|
)
|
|
prompts_after, pairs_after = self.snapshot()
|
|
changed_prompts = {*components, negative}
|
|
changed_pairs = {*active, archived}
|
|
for table_before, table_after, changed in (
|
|
(prompts_before, prompts_after, changed_prompts),
|
|
(pairs_before, pairs_after, changed_pairs),
|
|
):
|
|
self.assertNotIn("Scenery", [r["category"] for r in table_after.values()])
|
|
self.assertEqual(set(table_after), set(table_before))
|
|
for row_id, row in table_after.items():
|
|
old = table_before[row_id]
|
|
if row_id in changed:
|
|
self.assertEqual(row["category"], "Landscape")
|
|
self.assertNotEqual(row["updated_at"], _OLD_STAMP)
|
|
expected = {**old, "category": "Landscape"}
|
|
expected["updated_at"] = row["updated_at"]
|
|
self.assertEqual(row, expected)
|
|
else:
|
|
self.assertEqual(row, old)
|
|
self.assertEqual(prompts_after[variant]["category"], " Scenery ")
|
|
self.assertEqual(prompts_after[bystander]["category"], "Other")
|
|
for pair_id in active:
|
|
pair = self.db.get_prompt_pair(pair_id)
|
|
self.assertEqual(pair["category"], "Landscape")
|
|
self.assertEqual(len(pair["tags"]), 1)
|
|
self.assertEqual(pairs_after[archived]["archived"], 1)
|
|
self.assertEqual(pairs_after[archived]["title"], "Kept")
|
|
# Consolidating into an existing value reports it.
|
|
merge = self.preview(source=" Scenery ", target="Other")
|
|
self.assertTrue(merge["target_exists"])
|
|
receipt = self.apply(merge["impact_token"], source=" Scenery ", target="Other")
|
|
self.assertTrue(receipt["consolidated"])
|
|
self.assertEqual(
|
|
receipt["impact"],
|
|
{
|
|
"positive_prompts": 1,
|
|
"other_prompts": 0,
|
|
"negative_variants": 0,
|
|
"archived_negative_variants": 0,
|
|
},
|
|
)
|
|
self.assertEqual(self.rows("prompts")[variant]["category"], "Other")
|
|
|
|
def test_token_changes_with_membership_eligibility_archive_and_target_presence(
|
|
self,
|
|
):
|
|
components, negative, active, _archived = self.owners()
|
|
|
|
def add_owner():
|
|
self.save("Late owner", category="Scenery")
|
|
|
|
def archive_pair():
|
|
self.db.archive_prompt_pair(active[0])
|
|
|
|
def promote_negative():
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
|
|
(negative, "Promoted"),
|
|
)
|
|
|
|
def create_target():
|
|
self.save("Target owner", category="Landscape")
|
|
|
|
for mutate in (add_owner, archive_pair, promote_negative, create_target):
|
|
with self.subTest(mutate.__name__):
|
|
token = self.preview()["impact_token"]
|
|
mutate()
|
|
before = self.snapshot()
|
|
changes = self.db.model.get_connection().total_changes
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.apply(token)
|
|
self.assertEqual(caught.exception.code, "vocabulary-conflict")
|
|
self.assertEqual(caught.exception.epoch, self.epoch)
|
|
self.assertEqual(self.db.model.get_connection().total_changes, changes)
|
|
self.assertEqual(self.snapshot(), before)
|
|
token = self.preview()["impact_token"]
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"UPDATE prompts SET notes='edited' WHERE id=?", (components[0],)
|
|
)
|
|
self.assertTrue(self.apply(token)["consolidated"])
|
|
self.assertEqual(self.rows("prompts")[components[0]]["category"], "Landscape")
|
|
|
|
def test_category_command_guards(self):
|
|
self.owners()
|
|
before = self.snapshot()
|
|
stale = "00000000-0000-4000-8000-000000000000"
|
|
token = self.preview()["impact_token"]
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.command(
|
|
"preview_vocabulary_category",
|
|
expected_epoch=stale,
|
|
source="Scenery",
|
|
target="Landscape",
|
|
)
|
|
self.assertEqual(caught.exception.code, "storage-changed")
|
|
self.assertEqual(caught.exception.epoch, self.epoch)
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.command(
|
|
"apply_vocabulary_category",
|
|
expected_epoch=stale,
|
|
source="Scenery",
|
|
target="Landscape",
|
|
impact_token=token,
|
|
)
|
|
self.assertEqual(caught.exception.code, "storage-changed")
|
|
for source in ("Missing", "scenery", " Scenery"):
|
|
with self.assertRaises(LibraryCommandError) as caught:
|
|
self.preview(source=source)
|
|
self.assertEqual(caught.exception.code, "vocabulary-conflict")
|
|
for source, target in (
|
|
("Scenery", ""),
|
|
("Scenery", " "),
|
|
("Scenery", "x" * 101),
|
|
("Scenery", "bad\x01"),
|
|
("Scenery", "Scenery"),
|
|
("Scenery", " Scenery "),
|
|
("Scenery", None),
|
|
("Scenery", 7),
|
|
(None, "Landscape"),
|
|
("", "Landscape"),
|
|
(" ", "Landscape"),
|
|
):
|
|
with self.assertRaises(ValueError, msg=(source, target)):
|
|
self.preview(source=source, target=target)
|
|
for bad in ("", "A" * 64, token[:-1], None, 7):
|
|
with self.assertRaises(ValueError):
|
|
self.apply(bad)
|
|
self.assertEqual(self.snapshot(), before)
|
|
|
|
def test_category_apply_failure_rolls_back(self):
|
|
self.owners()
|
|
token = self.preview()["impact_token"]
|
|
with self.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"CREATE TRIGGER abort_pairs BEFORE UPDATE ON prompt_pairs "
|
|
"BEGIN SELECT RAISE(ABORT, 'boom'); END"
|
|
)
|
|
before = self.snapshot()
|
|
with self.assertRaises(sqlite3.Error):
|
|
self.apply(token)
|
|
self.assertEqual(self.snapshot(), before)
|
|
self.assertIn("Scenery", [r["category"] for r in before[0].values()])
|
|
self.assertFalse(self.db.model.get_connection().in_transaction)
|
|
|
|
|
|
class TestSharedTokenHashing(unittest.TestCase):
|
|
def test_both_tokens_keep_their_previous_values(self):
|
|
from database.library_tokens import library_edit_token
|
|
from database.vocabulary import category_impact_token
|
|
|
|
record = {
|
|
"id": 3,
|
|
"text": "t",
|
|
"title": None,
|
|
"category": "c",
|
|
"notes": "",
|
|
"rating": 0,
|
|
"tags": ["b", "a", "a"],
|
|
}
|
|
self.assertEqual(
|
|
library_edit_token("e", record),
|
|
"32534d4c92e4e9a7d8e31c30ecdbb07563703b5bd7ed216a8d2066c0dc44332b",
|
|
)
|
|
self.assertEqual(
|
|
category_impact_token(
|
|
"e", "S", "T", True, [(1, True), (2, False)], [(5, False)]
|
|
),
|
|
"cefe7f6a5544b7f6a203f5527810899c6af568aa7abc1b75adffbfb5ae6251ae",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|