Vito
caa9649f32
* feat(db): add pair schema and stage restore migrations for #193 * feat(database): persist reusable ordered prompt pairs (#193) * feat(api): expose reusable pairs and atomic metadata enrichment (#193) * feat(metadata): resolve canonical prompt pairs with bounded provenance (#193) * feat(scan): enrich image pairs without guessing or duplicate links (#193) * fix(database): preserve pair tags during global maintenance (#193) * feat(tracking): enrich live image pairs and cancel scans safely (#193) * feat(nodes): encode paired prompts and count cached recipe runs (#193) * fix(database): validate restored pair ownership and provenance (#193) * feat(admin): report pair scan outcomes and cancellation (#193) * feat(ui): add paired prompt editor and explicit recipe library for #193 * fix(autocomplete): include reusable pair metadata vocabulary (#193) * fix(pairs): honor recent sorting and conservative scan outcomes (#193) * fix(ui): protect paired draft saves and clarify negative variants (#193) * fix(metadata): preserve bounded paired execution evidence (#193) * fix(tracking): follow actual pair output roles and prefer execution evidence (#193) * docs(pairs): document live validation and retain renderer screenshots (#193) * test(api): inject disposable databases before API startup (#194) * test(api): canonicalize portable temporary gallery roots (#194) * docs(ui): prototype branching negative variant history (#193) * docs(ui): show selected negative beneath positive in timeline concept (#193) * docs: retain approved timeline for later gallery redesign (#193) * test(db): close restore fixture connections before cleanup (#194) * test(db): reuse existing handle guards in all pair fixtures (#194) * chore: clarify rollback test and intentional exception handling (#193)
62 lines
2.3 KiB
Python
62 lines
2.3 KiB
Python
"""Dependency checks exclude competing pair writers until mutation finishes."""
|
|
|
|
import os
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest.mock
|
|
|
|
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.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()
|
|
assert_closed(self, directory)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|