diff --git a/TESTING.md b/TESTING.md index e77e01e..86931b7 100644 --- a/TESTING.md +++ b/TESTING.md @@ -1589,6 +1589,27 @@ inspection's own stable stat (`tests/test_api_workspace_images.py` browser fixtures clear `media_image_tags`, `media_annotation_bytes` and `media_annotation_files` on reset and never the metadata row. +Review queue schema (#249, `database/review_schema.py`, +`tests/test_review_schema.py`): the group `media_review_metadata` (one row: +schema version, `review_index_revision`), `media_review_items` (one ambiguous +resolver result per `(byte_id, resolver_version, evidence_hash)`, citing an +annotation byte record `ON DELETE RESTRICT`, with 0..16 candidates, `limited`, +`state` pending/selected/unresolved and a revision) and `media_review_decisions` +(one row per item decision revision; `selected` exactly when a +`candidate_hash` is present; `pair_id`/`image_id` navigation is set to NULL, +never cascaded, when the pair or image is deleted, through +`idx_media_review_decisions_pair`/`_image`). Evidence is canonical JSON +(`canonical_evidence`: one object, sorted keys, no whitespace, ASCII only, no +NaN/infinity, at most 8 MiB) and `evidence_hash` is its SHA-256. The group is +valid only beside a valid annotation group (a review group without one is +refused before the annotation install), installs like the annotation group +(optional in old backups, required once current under `require_library_state`, +SAVEPOINT install validated before and after), and validation re-canonicalizes +every evidence document in one streamed pass at startup and on restore (about +0.6 s for 4,000 items plus 4,000 decisions of 3 KB). The browser fixtures clear +`media_review_decisions` and `media_review_items` before the annotation tables, +never the metadata row. + Capture store (#249, `utils/media_capture.py`, `tests/test_media_capture.py`): one explicit action captures one rooted media file into an owner-only `prompt-manager-captures-*` temp directory, streaming it in 1 MiB chunks while diff --git a/database/models.py b/database/models.py index 5268577..ce439dd 100644 --- a/database/models.py +++ b/database/models.py @@ -18,6 +18,7 @@ from typing import Optional, Set, Tuple from .library_schema import library_schema_error, migrate_library_state from .image_favorites import image_favorites_schema_error, migrate_image_favorites from .annotation_schema import annotation_schema_error, migrate_annotation_state +from .review_schema import migrate_review_state, review_schema_error from .saved_views import migrate_saved_views, saved_views_schema_error from .execution_history_schema import ( history_schema_error, @@ -303,7 +304,13 @@ class PromptModel: migrate_library_state(conn) migrate_image_favorites(conn) migrate_saved_views(conn) + # A review group without its annotation group is refused before the + # annotation install could make it look valid. + review_error = review_schema_error(conn) + if review_error: + raise sqlite3.DatabaseError(review_error) migrate_annotation_state(conn) + migrate_review_state(conn) def _migrate_prompt_pairs(self, conn: sqlite3.Connection) -> None: conn.execute("SAVEPOINT prompt_pair_schema") @@ -1023,9 +1030,9 @@ class PromptModel: Args: path: File to inspect require_library_state: Also require the tables every current - database has (Library state, image favorites, saved views - and image annotations), as a migrated restore stage must; - old backups may lack them. + database has (Library state, image favorites, saved views, + image annotations and the review queue), as a migrated + restore stage must; old backups may lack them. Returns: (ok, reason): reason is "ok" when the file is acceptable, otherwise @@ -1095,6 +1102,11 @@ class PromptModel: ) if annotation_error: return False, annotation_error + review_error = review_schema_error( + conn, allow_missing=not require_library_state + ) + if review_error: + return False, review_error return True, "ok" except sqlite3.Error as e: return False, f"Invalid SQLite database: {e}" diff --git a/database/review_schema.py b/database/review_schema.py new file mode 100644 index 0000000..e311448 --- /dev/null +++ b/database/review_schema.py @@ -0,0 +1,410 @@ +"""Needs review schema group (#249): capture-bound ambiguous items and decisions. + +The group installs like ``image_annotations``: optional in old backups, +required once a database is current, and valid only beside a valid image +annotation group, because every item cites one annotation byte record. It owns + +- ``media_review_metadata``: one row, the group's schema version and the + review index revision (moved by every queue change); +- ``media_review_items``: one ambiguous resolver result for one byte record, + keyed by ``(byte_id, resolver_version, evidence_hash)``, with its canonical + evidence, candidate count, ``limited`` flag, state and revision; +- ``media_review_decisions``: the auditable decision history of an item, one + row per decision revision, with optional navigation to the pair and image + the decision linked (nulled, never cascaded, when those are deleted). + +Evidence is canonical JSON (:func:`canonical_evidence`): one object, sorted +keys, no whitespace, ASCII only (so characters, bytes and the 8 MiB bound +agree and lone surrogates survive), no NaN or infinity. ``evidence_hash`` is +the SHA-256 of the item's evidence text. + +Installation never backfills anything and never touches the filesystem. +Validation reads only: catalog objects compared after whitespace compaction, +columns and foreign keys through the PRAGMAs, persisted values in single +bounded SQL passes, then one streamed pass that re-canonicalizes each evidence +document (one row in memory at a time). +""" + +import hashlib +import json +import re +import sqlite3 +from typing import Any, Optional + +from .annotation_schema import BYTES_TABLE, annotation_schema_error + +SCHEMA_VERSION = 1 +METADATA_TABLE = "media_review_metadata" +ITEMS_TABLE = "media_review_items" +DECISIONS_TABLE = "media_review_decisions" +TABLES = (METADATA_TABLE, ITEMS_TABLE, DECISIONS_TABLE) +MAX_EVIDENCE_BYTES = 8 * 1024 * 1024 +MAX_CANDIDATES = 16 +ITEM_STATES = ("pending", "selected", "unresolved") +DECISION_OUTCOMES = ("selected", "unresolved") +_TIMESTAMP_BYTES = 64 +_HEX64 = "length({c})=64 AND {c} NOT GLOB '*[^0-9a-f]*'" +_EVIDENCE_CHECK = f"CHECK(length(CAST(evidence AS BLOB)) <= {MAX_EVIDENCE_BYTES})" + +_TABLE_DDL = { + METADATA_TABLE: ( + f"CREATE TABLE {METADATA_TABLE} (singleton_id INTEGER PRIMARY KEY " + "CHECK(singleton_id=1), schema_version INTEGER NOT NULL " + f"CHECK(schema_version={SCHEMA_VERSION}), " + "review_index_revision INTEGER NOT NULL DEFAULT 0 " + "CHECK(review_index_revision>=0))" + ), + ITEMS_TABLE: ( + f"CREATE TABLE {ITEMS_TABLE} (id TEXT PRIMARY KEY CHECK(length(id)=36), " + f"byte_id TEXT NOT NULL REFERENCES {BYTES_TABLE}(id) ON DELETE RESTRICT, " + "resolver_version INTEGER NOT NULL, evidence_hash TEXT NOT NULL " + f"CHECK({_HEX64.format(c='evidence_hash')}), " + f"evidence TEXT NOT NULL {_EVIDENCE_CHECK}, " + "candidate_count INTEGER NOT NULL " + f"CHECK(candidate_count BETWEEN 0 AND {MAX_CANDIDATES}), " + "limited INTEGER NOT NULL CHECK(limited IN (0,1)), " + "state TEXT NOT NULL CHECK(state IN ('pending','selected','unresolved')), " + "revision INTEGER NOT NULL CHECK(revision>=0), created_at TEXT NOT NULL, " + "updated_at TEXT NOT NULL, UNIQUE(byte_id, resolver_version, evidence_hash))" + ), + DECISIONS_TABLE: ( + f"CREATE TABLE {DECISIONS_TABLE} (id TEXT PRIMARY KEY " + "CHECK(length(id)=36), item_id TEXT NOT NULL " + f"REFERENCES {ITEMS_TABLE}(id) ON DELETE RESTRICT, " + "decision_revision INTEGER NOT NULL, created_at TEXT NOT NULL, " + "outcome TEXT NOT NULL CHECK(outcome IN ('selected','unresolved')), " + "candidate_hash TEXT CHECK(candidate_hash IS NULL OR " + f"({_HEX64.format(c='candidate_hash')})), " + f"evidence TEXT NOT NULL {_EVIDENCE_CHECK}, " + "pair_id INTEGER REFERENCES prompt_pairs(id) ON DELETE SET NULL, " + "image_id INTEGER REFERENCES generated_images(id) ON DELETE SET NULL, " + "UNIQUE(item_id, decision_revision), " + "CHECK((outcome='selected') = (candidate_hash IS NOT NULL)))" + ), +} +_INDEX_DDL = { + # The queue reads pending items oldest first with a stable tiebreak. + "idx_media_review_items_queue": ( + "CREATE INDEX idx_media_review_items_queue ON " + f"{ITEMS_TABLE}(state, created_at, id)" + ), + # Deleting a pair or an image nulls decision navigation by index, not scan. + "idx_media_review_decisions_pair": ( + "CREATE INDEX idx_media_review_decisions_pair ON " f"{DECISIONS_TABLE}(pair_id)" + ), + "idx_media_review_decisions_image": ( + "CREATE INDEX idx_media_review_decisions_image ON " + f"{DECISIONS_TABLE}(image_id)" + ), +} +# ``sqlite_autoindex__``: the TEXT primary key and the UNIQUE key. +_AUTOINDEXES = {ITEMS_TABLE: 2, DECISIONS_TABLE: 2} +# (name, declared type, not null, default, primary key position) +_COLUMNS = { + METADATA_TABLE: ( + ("singleton_id", "INTEGER", 0, None, 1), + ("schema_version", "INTEGER", 1, None, 0), + ("review_index_revision", "INTEGER", 1, "0", 0), + ), + ITEMS_TABLE: ( + ("id", "TEXT", 0, None, 1), + ("byte_id", "TEXT", 1, None, 0), + ("resolver_version", "INTEGER", 1, None, 0), + ("evidence_hash", "TEXT", 1, None, 0), + ("evidence", "TEXT", 1, None, 0), + ("candidate_count", "INTEGER", 1, None, 0), + ("limited", "INTEGER", 1, None, 0), + ("state", "TEXT", 1, None, 0), + ("revision", "INTEGER", 1, None, 0), + ("created_at", "TEXT", 1, None, 0), + ("updated_at", "TEXT", 1, None, 0), + ), + DECISIONS_TABLE: ( + ("id", "TEXT", 0, None, 1), + ("item_id", "TEXT", 1, None, 0), + ("decision_revision", "INTEGER", 1, None, 0), + ("created_at", "TEXT", 1, None, 0), + ("outcome", "TEXT", 1, None, 0), + ("candidate_hash", "TEXT", 0, None, 0), + ("evidence", "TEXT", 1, None, 0), + ("pair_id", "INTEGER", 0, None, 0), + ("image_id", "INTEGER", 0, None, 0), + ), +} +# (column, parent table, parent column, on delete) +_FOREIGN_KEYS = { + METADATA_TABLE: set(), + ITEMS_TABLE: {("byte_id", BYTES_TABLE, "id", "RESTRICT")}, + DECISIONS_TABLE: { + ("item_id", ITEMS_TABLE, "id", "RESTRICT"), + ("pair_id", "prompt_pairs", "id", "SET NULL"), + ("image_id", "generated_images", "id", "SET NULL"), + }, +} + + +class _NonCanonical(ValueError): + pass + + +def _reject_constant(name: str) -> Any: + raise _NonCanonical(f"JSON constant {name} is not allowed") + + +def canonical_evidence(value: Any) -> str: + """Serialize one evidence object canonically; ``ValueError`` when invalid. + + The text is a JSON object with sorted keys, no whitespace, ASCII only and + no NaN/infinity, at most ``MAX_EVIDENCE_BYTES`` bytes. + """ + if not isinstance(value, dict): + raise ValueError("Review evidence must be a JSON object") + try: + text = json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + allow_nan=False, + ) + except (TypeError, RecursionError) as error: + raise ValueError(f"Review evidence is not serializable: {error}") from None + if len(text) > MAX_EVIDENCE_BYTES: + raise ValueError("Review evidence is larger than 8 MiB") + return text + + +def evidence_digest(text: str) -> str: + """SHA-256 of canonical evidence text (ASCII, hence UTF-8 identical).""" + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def _is_canonical(text: Any) -> bool: + if not isinstance(text, str): + return False + try: + value = json.loads(text, parse_constant=_reject_constant) + return canonical_evidence(value) == text + except (ValueError, RecursionError): + return False + + +def _compact(sql: str) -> str: + return re.sub(r"\s+", "", sql).lower() + + +def _uuid(column: str) -> str: + """SQL: *column* is a lowercase hyphenated 36-character UUID string.""" + return ( + f"typeof({column}) = 'text' AND length({column}) = 36 " + f"AND substr({column}, 9, 1) = '-' AND substr({column}, 14, 1) = '-' " + f"AND substr({column}, 19, 1) = '-' AND substr({column}, 24, 1) = '-' " + f"AND length(replace({column}, '-', '')) = 32 " + f"AND replace({column}, '-', '') NOT GLOB '*[^0-9a-f]*'" + ) + + +def _timestamp(column: str) -> str: + return ( + f"typeof({column}) = 'text' " + f"AND length(CAST({column} AS BLOB)) <= {_TIMESTAMP_BYTES}" + ) + + +def _non_negative(column: str) -> str: + return f"typeof({column}) = 'integer' AND {column} >= 0" + + +def _hex64(column: str) -> str: + return f"typeof({column}) = 'text' AND " + _HEX64.format(c=column) + + +def _evidence(column: str) -> str: + # Printable ASCII only is a cheap necessary condition for canonical text; + # the streamed pass below proves the rest. + return ( + f"typeof({column}) = 'text' " + f"AND length(CAST({column} AS BLOB)) BETWEEN 2 AND {MAX_EVIDENCE_BYTES} " + f"AND CAST({column} AS BLOB) NOT GLOB '*[^ -~]*'" + ) + + +def _nullable_integer(column: str) -> str: + return f"({column} IS NULL OR typeof({column}) = 'integer')" + + +def _in(column: str, values: tuple) -> str: + return f"typeof({column}) = 'text' AND {column} IN ({repr(values)[1:-1]})" + + +# Every row of a table must satisfy its predicate; one pass each, LIMIT 1. +_ROW_PREDICATES = { + ITEMS_TABLE: " AND ".join( + ( + _uuid("id"), + _uuid("byte_id"), + "typeof(resolver_version) = 'integer'", + _hex64("evidence_hash"), + _evidence("evidence"), + _non_negative("candidate_count"), + f"candidate_count <= {MAX_CANDIDATES}", + "typeof(limited) = 'integer' AND limited IN (0, 1)", + _in("state", ITEM_STATES), + _non_negative("revision"), + _timestamp("created_at"), + _timestamp("updated_at"), + ) + ), + DECISIONS_TABLE: " AND ".join( + ( + _uuid("id"), + _uuid("item_id"), + _non_negative("decision_revision"), + _timestamp("created_at"), + _in("outcome", DECISION_OUTCOMES), + "(CASE WHEN outcome = 'selected' THEN " + f"{_hex64('candidate_hash')} ELSE candidate_hash IS NULL END)", + _evidence("evidence"), + _nullable_integer("pair_id"), + _nullable_integer("image_id"), + ) + ), +} + + +def _expected_catalog() -> set: + objects = {("table", name, _compact(sql)) for name, sql in _TABLE_DDL.items()} + objects |= {("index", name, _compact(sql)) for name, sql in _INDEX_DDL.items()} + objects |= { + ("index", f"sqlite_autoindex_{table}_{n}", "") + for table, count in _AUTOINDEXES.items() + for n in range(1, count + 1) + } + return objects + + +def _catalog(conn: sqlite3.Connection) -> list: + names = TABLES + tuple(_INDEX_DDL) + marks = ",".join("?" * len(names)) + table_marks = ",".join("?" * len(TABLES)) + return conn.execute( # only "?" placeholders are interpolated + "SELECT type,name,sql FROM sqlite_master " + f"WHERE name IN ({marks}) OR tbl_name IN ({table_marks})", # nosec B608 + names + TABLES, + ).fetchall() + + +def _structure_error(conn: sqlite3.Connection) -> Optional[str]: + for table, expected in _COLUMNS.items(): + columns = tuple( + (row[1], row[2].upper(), row[3], row[4], row[5]) + for row in conn.execute(f"PRAGMA table_info({table})") + ) + if columns != expected: + return f"Review queue table {table} has invalid columns" + foreign_keys = { + (row[3], row[2], row[4], row[6]) + for row in conn.execute(f"PRAGMA foreign_key_list({table})") + } + if foreign_keys != _FOREIGN_KEYS[table]: + return f"Review queue table {table} has invalid foreign keys" + if conn.execute(f"PRAGMA foreign_key_check({table})").fetchone(): + return f"Review queue table {table} contains foreign key violations" + return None + + +def _evidence_error(conn: sqlite3.Connection) -> Optional[str]: + """Stream every evidence document once; one row is held at a time.""" + for evidence_hash, text in conn.execute( + f"SELECT evidence_hash, evidence FROM {ITEMS_TABLE}" # nosec B608 + ): + if not _is_canonical(text) or evidence_digest(text) != evidence_hash: + return f"Review queue table {ITEMS_TABLE} contains invalid evidence" + for (text,) in conn.execute( + f"SELECT evidence FROM {DECISIONS_TABLE}" # nosec B608 + ): + if not _is_canonical(text): + return f"Review queue table {DECISIONS_TABLE} contains invalid evidence" + return None + + +def _value_error(conn: sqlite3.Connection) -> Optional[str]: + # Only this module's constants are interpolated into the SQL below. + rows = conn.execute( + "SELECT COUNT(*), MIN(typeof(singleton_id) = 'integer' " # nosec B608 + "AND singleton_id = 1 AND typeof(schema_version) = 'integer' " + f"AND schema_version = {SCHEMA_VERSION} " + f"AND {_non_negative('review_index_revision')}) FROM {METADATA_TABLE}" + ).fetchone() + if rows[0] != 1 or rows[1] != 1: + return "Review queue metadata is invalid" + for table, predicate in _ROW_PREDICATES.items(): + if conn.execute( + f"SELECT 1 FROM {table} WHERE NOT ({predicate}) LIMIT 1" # nosec B608 + ).fetchone(): + return f"Review queue table {table} contains invalid values" + return _evidence_error(conn) + + +def review_schema_error( + conn: sqlite3.Connection, *, allow_missing: bool = True +) -> Optional[str]: + """Reject partial or unsupported review schemas and invalid rows. + + Reads only: nothing is ever repaired. A database without any of the + group's objects is acceptable for backups that predate it unless + ``allow_missing`` is false. A present group requires a present, valid + image annotation group. + """ + objects = _catalog(conn) + if not objects: + return None if allow_missing else "Missing review queue schema" + annotation_error = annotation_schema_error(conn, allow_missing=False) + if annotation_error: + return ( + "Review queue schema requires a valid image annotation schema: " + + annotation_error + ) + present = {name for kind, name, _sql in objects if kind == "table"} + if present != set(TABLES): + return "Incomplete review queue schema" + actual = {(kind, name, _compact(sql or "")) for kind, name, sql in objects} + if actual != _expected_catalog(): + return "Review queue has an invalid schema definition" + return _structure_error(conn) or _value_error(conn) + + +def migrate_review_state(conn: sqlite3.Connection) -> None: + """Install the group atomically; validate before and after creating it. + + Runs after ``migrate_annotation_state``: the group is only valid beside + the annotation group it cites. + """ + conn.execute("SAVEPOINT media_review_schema") + try: + installed = bool(_catalog(conn)) + error = review_schema_error(conn) + if error: + raise sqlite3.DatabaseError(error) + if installed: + # Already complete and valid: nothing to create, and the evidence + # pass is the expensive part of validation, so it runs once. + conn.execute("RELEASE media_review_schema") + return + for sql in _TABLE_DDL.values(): + conn.execute(sql.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)) + for sql in _INDEX_DDL.values(): + conn.execute(sql.replace("INDEX", "INDEX IF NOT EXISTS", 1)) + conn.execute( + f"INSERT OR IGNORE INTO {METADATA_TABLE} " + "(singleton_id, schema_version, review_index_revision) VALUES (1, ?, 0)", + (SCHEMA_VERSION,), + ) + error = review_schema_error(conn, allow_missing=False) + if error: + raise sqlite3.DatabaseError(error) + conn.execute("RELEASE media_review_schema") + except BaseException: + conn.execute("ROLLBACK TO media_review_schema") + conn.execute("RELEASE media_review_schema") + raise diff --git a/tests/browser/server.py b/tests/browser/server.py index 4ff976d..6cf7fe5 100644 --- a/tests/browser/server.py +++ b/tests/browser/server.py @@ -219,6 +219,8 @@ def create_app(root: Path) -> web.Application: for table in ( "saved_views", "image_favorites", + "media_review_decisions", + "media_review_items", "media_image_tags", "media_annotation_bytes", "media_annotation_files", diff --git a/tests/browser/workspace_concept_fixture.py b/tests/browser/workspace_concept_fixture.py index b22c362..1b1b2fc 100644 --- a/tests/browser/workspace_concept_fixture.py +++ b/tests/browser/workspace_concept_fixture.py @@ -173,6 +173,8 @@ _TABLES = ( "execution_context_texts", "saved_views", "image_favorites", + "media_review_decisions", + "media_review_items", "media_image_tags", "media_annotation_bytes", "media_annotation_files", diff --git a/tests/browser/workspace_library_editor_fixture.py b/tests/browser/workspace_library_editor_fixture.py index 7b852a1..ab375c8 100644 --- a/tests/browser/workspace_library_editor_fixture.py +++ b/tests/browser/workspace_library_editor_fixture.py @@ -28,6 +28,8 @@ EDITOR_METADATA = { _TABLES = ( "saved_views", "image_favorites", + "media_review_decisions", + "media_review_items", "media_image_tags", "media_annotation_bytes", "media_annotation_files", diff --git a/tests/browser/workspace_tagging_fixture.py b/tests/browser/workspace_tagging_fixture.py index 31cbd2f..9bf0214 100644 --- a/tests/browser/workspace_tagging_fixture.py +++ b/tests/browser/workspace_tagging_fixture.py @@ -44,6 +44,8 @@ MAX_EXTRA_IMAGES = 160 _TABLES = ( "saved_views", "image_favorites", + "media_review_decisions", + "media_review_items", "media_image_tags", "media_annotation_bytes", "media_annotation_files", diff --git a/tests/browser/workspace_visual_fixture.py b/tests/browser/workspace_visual_fixture.py index 9a7cdad..3a98342 100644 --- a/tests/browser/workspace_visual_fixture.py +++ b/tests/browser/workspace_visual_fixture.py @@ -114,6 +114,8 @@ def seed_workspace_visual(api: PromptManagerAPI, root: Path) -> None: for table in ( "saved_views", "image_favorites", + "media_review_decisions", + "media_review_items", "media_image_tags", "media_annotation_bytes", "media_annotation_files", diff --git a/tests/browser/workspace_vocabulary_fixture.py b/tests/browser/workspace_vocabulary_fixture.py index 69e0fb9..10c5a8c 100644 --- a/tests/browser/workspace_vocabulary_fixture.py +++ b/tests/browser/workspace_vocabulary_fixture.py @@ -81,6 +81,8 @@ MAX_EXTRA_CATEGORIES = 200 _TABLES = ( "saved_views", "image_favorites", + "media_review_decisions", + "media_review_items", "media_image_tags", "media_annotation_bytes", "media_annotation_files", diff --git a/tests/test_annotation_schema.py b/tests/test_annotation_schema.py index cc37b9a..4961117 100644 --- a/tests/test_annotation_schema.py +++ b/tests/test_annotation_schema.py @@ -30,6 +30,11 @@ FILES = "media_annotation_files" BYTES = "media_annotation_bytes" IMAGE_TAGS = "media_image_tags" TABLES = (METADATA, FILES, BYTES, IMAGE_TAGS) +REVIEW_TABLES = ( + "media_review_decisions", + "media_review_items", + "media_review_metadata", +) # The exact DDL the plan pins; the module must produce precisely these objects. DDL = { METADATA: ( @@ -202,10 +207,14 @@ class AnnotationSchemaTests(unittest.TestCase): ) def drop_group(self, path): - """Remove the group from a closed database file (an old backup).""" + """Remove the group from a closed database file (an old backup). + + The review queue group cites annotation bytes and predates nothing, so + an old backup lacks it as well. + """ with closing(sqlite3.connect(path)) as raw, raw: raw.execute("PRAGMA foreign_keys=OFF") - for table in reversed(TABLES): + for table in REVIEW_TABLES + tuple(reversed(TABLES)): raw.execute(f"DROP TABLE {table}") def test_fresh_install_and_reopen_is_idempotent(self): diff --git a/tests/test_database_backup.py b/tests/test_database_backup.py index deb1346..b8b56f5 100644 --- a/tests/test_database_backup.py +++ b/tests/test_database_backup.py @@ -454,6 +454,9 @@ class TestRestoreFromFile(BackupTestCase): def test_restore_old_backup_without_annotations(self): tables = ( + "media_review_decisions", + "media_review_items", + "media_review_metadata", "media_image_tags", "media_annotation_bytes", "media_annotation_files", @@ -490,6 +493,70 @@ class TestRestoreFromFile(BackupTestCase): 0, ) + def test_restore_annotations_era_backup_without_review_queue(self): + tables = ( + "media_review_decisions", + "media_review_items", + "media_review_metadata", + ) + src = self._make_source_db(["predates the review queue"]) + with closing(sqlite3.connect(src)) as conn: + for table in tables: + conn.execute(f"DROP TABLE {table}") + conn.commit() + self.assertTrue(PromptModel.verify_database_file(src)[0]) + ok, reason = PromptModel.verify_database_file(src, require_library_state=True) + self.assertFalse(ok) + self.assertEqual(reason, "Missing review queue schema") + + self.db.model.restore_from_file(src) + + self.assertEqual(_texts(self.path), ["predates the review queue"]) + with closing(sqlite3.connect(self.path)) as conn: + self.assertEqual( + conn.execute("SELECT * FROM media_review_metadata").fetchall(), + [(1, 1, 0)], + ) + self.assertEqual( + conn.execute("SELECT COUNT(*) FROM media_review_items").fetchone()[0], + 0, + ) + ok, reason = PromptModel.verify_database_file( + self.path, require_library_state=True + ) + self.assertTrue(ok, reason) + with closing(sqlite3.connect(src)) as conn: # the source is untouched + self.assertEqual( + conn.execute( + "SELECT COUNT(*) FROM sqlite_master WHERE name LIKE 'media_review%'" + ).fetchone()[0], + 0, + ) + + def test_restore_rejects_review_queue_without_annotations(self): + src = self._make_source_db(["review without annotations"]) + with closing(sqlite3.connect(src)) as conn: + conn.execute("PRAGMA foreign_keys=OFF") + for table in ( + "media_image_tags", + "media_annotation_bytes", + "media_annotation_files", + "media_annotation_metadata", + ): + conn.execute(f"DROP TABLE {table}") + conn.commit() + before = Path(src).read_bytes() + live_before = _texts(self.path) + ok, reason = PromptModel.verify_database_file(src) + self.assertFalse(ok) + self.assertIn("image annotation", reason) + + with self.assertRaises(ValueError): + self.db.model.restore_from_file(src) + + self.assertEqual(Path(src).read_bytes(), before) + self.assertEqual(_texts(self.path), live_before) + def test_restore_preserves_favorites_and_closes_handles(self): self.db.set_image_favorite("output", "live.png") with tempfile.TemporaryDirectory() as directory: diff --git a/tests/test_pair_schema_restore.py b/tests/test_pair_schema_restore.py index dff5420..24b1e87 100644 --- a/tests/test_pair_schema_restore.py +++ b/tests/test_pair_schema_restore.py @@ -258,6 +258,10 @@ class PairSchemaTests(unittest.TestCase): ], [(1, 1, 0)], ) + self.assertEqual( + [tuple(r) for r in conn.execute("SELECT * FROM media_review_metadata")], + [(1, 1, 0)], + ) self.assertTrue(PromptModel.verify_database_file(live.db_path)[0]) self.assertTrue( PromptModel.verify_database_file(live.db_path, require_library_state=True)[ diff --git a/tests/test_review_schema.py b/tests/test_review_schema.py new file mode 100644 index 0000000..674ced9 --- /dev/null +++ b/tests/test_review_schema.py @@ -0,0 +1,870 @@ +"""Needs review schema group (#249): install, validation and restore safety.""" + +import hashlib +import json +import re +import sqlite3 +import tempfile +import unittest +from contextlib import closing +from pathlib import Path +from unittest.mock import patch + +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.models import PromptModel +from database.operations import PromptDatabase + +try: # guarded so a missing module fails assertions, not test loading + from database.review_schema import ( + canonical_evidence, + evidence_digest, + migrate_review_state, + review_schema_error, + ) +except ImportError: + canonical_evidence = None + evidence_digest = None + migrate_review_state = None + review_schema_error = None + +METADATA = "media_review_metadata" +ITEMS = "media_review_items" +DECISIONS = "media_review_decisions" +TABLES = (METADATA, ITEMS, DECISIONS) +ANNOTATION_TABLES = ( + "media_annotation_metadata", + "media_annotation_files", + "media_annotation_bytes", + "media_image_tags", +) +MAX_EVIDENCE_BYTES = 8 * 1024 * 1024 +# The exact DDL the plan pins; the module must produce precisely these objects. +DDL = { + METADATA: ( + f"CREATE TABLE {METADATA}(singleton_id INTEGER PRIMARY KEY " + "CHECK(singleton_id=1), schema_version INTEGER NOT NULL " + "CHECK(schema_version=1), review_index_revision INTEGER NOT NULL " + "DEFAULT 0 CHECK(review_index_revision>=0))" + ), + ITEMS: ( + f"CREATE TABLE {ITEMS}(id TEXT PRIMARY KEY CHECK(length(id)=36), " + "byte_id TEXT NOT NULL REFERENCES media_annotation_bytes(id) " + "ON DELETE RESTRICT, resolver_version INTEGER NOT NULL, " + "evidence_hash TEXT NOT NULL CHECK(length(evidence_hash)=64 AND " + "evidence_hash NOT GLOB '*[^0-9a-f]*'), evidence TEXT NOT NULL " + f"CHECK(length(CAST(evidence AS BLOB)) <= {MAX_EVIDENCE_BYTES}), " + "candidate_count INTEGER NOT NULL CHECK(candidate_count BETWEEN 0 AND 16), " + "limited INTEGER NOT NULL CHECK(limited IN (0,1)), state TEXT NOT NULL " + "CHECK(state IN ('pending','selected','unresolved')), revision INTEGER " + "NOT NULL CHECK(revision>=0), created_at TEXT NOT NULL, updated_at TEXT " + "NOT NULL, UNIQUE(byte_id, resolver_version, evidence_hash))" + ), + DECISIONS: ( + f"CREATE TABLE {DECISIONS}(id TEXT PRIMARY KEY CHECK(length(id)=36), " + f"item_id TEXT NOT NULL REFERENCES {ITEMS}(id) ON DELETE RESTRICT, " + "decision_revision INTEGER NOT NULL, created_at TEXT NOT NULL, " + "outcome TEXT NOT NULL CHECK(outcome IN ('selected','unresolved')), " + "candidate_hash TEXT CHECK(candidate_hash IS NULL OR " + "(length(candidate_hash)=64 AND candidate_hash NOT GLOB '*[^0-9a-f]*')), " + "evidence TEXT NOT NULL " + f"CHECK(length(CAST(evidence AS BLOB)) <= {MAX_EVIDENCE_BYTES}), " + "pair_id INTEGER REFERENCES prompt_pairs(id) ON DELETE SET NULL, " + "image_id INTEGER REFERENCES generated_images(id) ON DELETE SET NULL, " + "UNIQUE(item_id, decision_revision), " + "CHECK((outcome='selected') = (candidate_hash IS NOT NULL)))" + ), +} +INDEXES = { + "idx_media_review_items_queue": ( + f"CREATE INDEX idx_media_review_items_queue ON {ITEMS}" + "(state, created_at, id)" + ), + "idx_media_review_decisions_pair": ( + f"CREATE INDEX idx_media_review_decisions_pair ON {DECISIONS}(pair_id)" + ), + "idx_media_review_decisions_image": ( + f"CREATE INDEX idx_media_review_decisions_image ON {DECISIONS}(image_id)" + ), +} +AUTOINDEXES = {ITEMS: 2, DECISIONS: 2} +LEGACY_TABLES = ( + "prompts", + "generated_images", + "tags", + "prompt_tags", + "prompt_pairs", + "prompt_pair_tags", + "image_favorites", +) + ANNOTATION_TABLES +FILE_ID = "0f8e4b2a-1c3d-4e5f-8a9b-0c1d2e3f4a5b" +BYTE_ID = "6a1b2c3d-4e5f-4a6b-8c9d-0e1f2a3b4c5d" +ITEM_ID = "1d2e3f4a-5b6c-4d7e-8f9a-0b1c2d3e4f5a" +DECISION_ID = "2e3f4a5b-6c7d-4e8f-9a0b-1c2d3e4f5a6b" +OTHER_ID = "7b2c3d4e-5f6a-4b7c-9d0e-1f2a3b4c5d6e" +DIGEST = hashlib.sha256(b"img").hexdigest() +CANDIDATE = hashlib.sha256(b"candidate").hexdigest() +NOW = "2026-10-22T00:00:00.000000+00:00" +EVIDENCE = '{"candidates":[],"sha256":"' + DIGEST + '"}' +EVIDENCE_HASH = hashlib.sha256(EVIDENCE.encode("ascii")).hexdigest() + + +def _compact(sql): + return re.sub(r"\s+", "", sql or "").lower() + + +def expected_catalog(): + objects = {("table", name, _compact(sql)) for name, sql in DDL.items()} + objects |= {("index", name, _compact(sql)) for name, sql in INDEXES.items()} + objects |= { + ("index", f"sqlite_autoindex_{table}_{n}", "") + for table, count in AUTOINDEXES.items() + for n in range(1, count + 1) + } + return objects + + +def catalog(conn): + names = tuple(TABLES) + tuple(INDEXES) + marks = ",".join("?" * len(names)) + table_marks = ",".join("?" * len(TABLES)) + return { + (kind, name, _compact(sql)) + for kind, name, sql in conn.execute( + "SELECT type,name,sql FROM sqlite_master " + f"WHERE name IN ({marks}) OR tbl_name IN ({table_marks})", # nosec B608 + names + tuple(TABLES), + ) + } + + +def rows(conn, sql, *params): + """Plain tuples whatever row factory *conn* uses.""" + return [tuple(row) for row in conn.execute(sql, params)] + + +def metadata(conn): + return rows(conn, f"SELECT schema_version,review_index_revision FROM {METADATA}") + + +def table_rows(conn, tables): + found = {} + for table in tables: + present = conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() + if present: + found[table] = rows(conn, f"SELECT * FROM {table}") # nosec B608 + return found + + +def insert_byte(conn, file_id=FILE_ID, byte_id=BYTE_ID, locator="/out/img.png"): + conn.execute( + "INSERT INTO media_annotation_files VALUES(?,?,?)", (file_id, locator, NOW) + ) + conn.execute( + "INSERT INTO media_annotation_bytes VALUES(?,?,?,?,?,?,?,0)", + (byte_id, file_id, DIGEST, 3, 1_700_000_000_000_000_000, NOW, NOW), + ) + + +def insert_item(conn, *, item_id=ITEM_ID, evidence=EVIDENCE, candidates=2, **extra): + values = { + "id": item_id, + "byte_id": BYTE_ID, + "resolver_version": 1, + "evidence_hash": hashlib.sha256(evidence.encode("utf-8")).hexdigest(), + "evidence": evidence, + "candidate_count": candidates, + "limited": 0, + "state": "pending", + "revision": 0, + "created_at": NOW, + "updated_at": NOW, + } + values.update(extra) + columns = ",".join(values) + marks = ",".join("?" * len(values)) + conn.execute( + f"INSERT INTO {ITEMS}({columns}) VALUES({marks})", # nosec B608 + tuple(values.values()), + ) + + +def insert_decision(conn, *, decision_id=DECISION_ID, **extra): + values = { + "id": decision_id, + "item_id": ITEM_ID, + "decision_revision": 1, + "created_at": NOW, + "outcome": "selected", + "candidate_hash": CANDIDATE, + "evidence": EVIDENCE, + "pair_id": None, + "image_id": None, + } + values.update(extra) + columns = ",".join(values) + marks = ",".join("?" * len(values)) + conn.execute( + f"INSERT INTO {DECISIONS}({columns}) VALUES({marks})", # nosec B608 + tuple(values.values()), + ) + + +def insert_group_rows(conn, **item): + insert_byte(conn) + insert_item(conn, **item) + insert_decision(conn) + + +def populate_legacy(conn): + conn.execute( + "INSERT INTO prompts(id,text,hash,category,notes,run_count) VALUES" + "(7,'positive','p','keep','notes',12)" + ) + conn.execute("INSERT INTO prompts(id,text,hash) VALUES(9,'negative','n')") + conn.execute( + "INSERT INTO generated_images(id,prompt_id,image_path,filename,file_path) " + "VALUES(4,7,'/tmp/image.png','image.png','/tmp/image.png')" + ) + conn.execute("INSERT INTO tags(id,name) VALUES(3,'keep-tag')") + conn.execute("INSERT INTO prompt_tags VALUES(7,3)") + conn.execute( + "INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) " + "VALUES(6,7,9)" + ) + conn.execute("INSERT INTO prompt_pair_tags VALUES(6,3)") + conn.execute("INSERT INTO image_favorites VALUES('root','a.png',?)", (NOW,)) + + +def full_capacity_evidence(): + """Sixteen candidates of control, lone-surrogate, astral and combining + text, exactly 8 MiB.""" + text = '\x00\x07\x1f
\ud800 é é "q" \\ \U0001f9ea' + candidates = [ + { + "hash": hashlib.sha256(str(n).encode()).hexdigest(), + "negative": {"text": text * 3}, + "positive": {"text": text * 5}, + "sampler_id": str(n), + "state": "resolved", + } + for n in range(16) + ] + value = {"candidates": candidates, "pad": ""} + base = len(canonical_evidence(value).encode("utf-8")) + value["pad"] = "x" * (MAX_EVIDENCE_BYTES - base) + return value + + +class ReviewSchemaTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.addCleanup(assert_closed, self, self.temp.name) + self.addCleanup(PromptModel.reset_schema_cache) + self.models = [] + self.addCleanup(self.close_models) + self.path = str(Path(self.temp.name) / "live.db") + + def close_models(self): + for model in self.models: + model.close_all() + + def model(self, path=None): + model = PromptModel(path or self.path) + self.models.append(model) + return model + + def require_module(self): + self.assertIsNotNone( + review_schema_error, "database.review_schema is not implemented" + ) + + def drop_tables(self, path, tables=TABLES): + """Remove tables from a closed database file (an old backup).""" + with closing(sqlite3.connect(path)) as raw, raw: + raw.execute("PRAGMA foreign_keys=OFF") + for table in reversed(tables): + raw.execute(f"DROP TABLE {table}") + + def test_fresh_install_and_reopen_is_idempotent(self): + self.require_module() + model = self.model() + with model.get_connection() as conn: + self.assertEqual(catalog(conn), expected_catalog()) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + self.assertEqual(metadata(conn), [(1, 0)]) + insert_group_rows(conn) + snapshot = table_rows(conn, TABLES) + self.assertEqual( + PromptModel.verify_database_file(self.path, require_library_state=True), + (True, "ok"), + ) + model.close_all() + for _ in range(2): + PromptModel.reset_schema_cache() + model = self.model() + with model.get_connection() as conn: + self.assertEqual(catalog(conn), expected_catalog()) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + self.assertEqual(table_rows(conn, TABLES), snapshot) + # Running the migration again on a current database changes nothing. + migrate_review_state(conn) + self.assertEqual(table_rows(conn, TABLES), snapshot) + model.close_all() + # Items and decisions keep the byte record and the item they cite. + PromptModel.reset_schema_cache() + model = self.model() + with model.get_connection() as conn: + with self.assertRaises(sqlite3.IntegrityError): + conn.execute("DELETE FROM media_annotation_files") + with self.assertRaises(sqlite3.IntegrityError): + conn.execute(f"DELETE FROM {ITEMS}") + self.assertEqual(table_rows(conn, TABLES), snapshot) + + def test_old_database_without_group_migrates_without_touching_legacy_rows(self): + self.require_module() + for label, dropped in ( + ("annotations only", TABLES), + ("predates annotations", ANNOTATION_TABLES + TABLES), + ): + with self.subTest(label): + path = str(Path(self.temp.name) / f"{label.replace(' ', '-')}.db") + model = self.model(path) + with model.get_connection() as conn: + populate_legacy(conn) + if dropped == TABLES: + insert_byte(conn) + model.close_all() + self.drop_tables(path, dropped) + with closing(sqlite3.connect(path)) as raw: + self.assertIsNone(review_schema_error(raw)) + self.assertEqual( + review_schema_error(raw, allow_missing=False), + "Missing review queue schema", + ) + legacy_rows_before = table_rows(raw, LEGACY_TABLES) + self.assertTrue(PromptModel.verify_database_file(path)[0]) + self.assertFalse( + PromptModel.verify_database_file(path, require_library_state=True)[ + 0 + ] + ) + PromptModel.reset_schema_cache() + + model = self.model(path) + + with model.get_connection() as conn: + self.assertEqual(catalog(conn), expected_catalog()) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + self.assertEqual(metadata(conn), [(1, 0)]) + counts = [ + conn.execute(f"SELECT COUNT(*) FROM {t}").fetchone()[0] + for t in (ITEMS, DECISIONS) + ] + legacy_rows_after = { + table: value + for table, value in table_rows(conn, LEGACY_TABLES).items() + if table in legacy_rows_before + } + self.assertEqual(counts, [0, 0]) + self.assertEqual(legacy_rows_after, legacy_rows_before) + self.assertEqual( + PromptModel.verify_database_file(path, require_library_state=True), + (True, "ok"), + ) + model.close_all() + + def test_annotations_only_installation_is_valid(self): + self.require_module() + model = self.model() + with model.get_connection() as conn: + insert_byte(conn) + model.close_all() + self.drop_tables(self.path) + with closing(sqlite3.connect(self.path)) as raw: + self.assertIsNone(review_schema_error(raw)) + self.assertEqual(PromptModel.verify_database_file(self.path), (True, "ok")) + self.assertEqual( + PromptModel.verify_database_file(self.path, require_library_state=True), + (False, "Missing review queue schema"), + ) + + def test_review_group_without_annotation_group_is_invalid(self): + self.require_module() + model = self.model() + model.close_all() + self.drop_tables(self.path, ANNOTATION_TABLES) + with closing(sqlite3.connect(self.path)) as raw: + for allow_missing in (True, False): + error = review_schema_error(raw, allow_missing=allow_missing) + self.assertIsNotNone(error) + self.assertIn("image annotation", error.lower()) + self.assertFalse(PromptModel.verify_database_file(self.path)[0]) + PromptModel.reset_schema_cache() + with self.assertRaises(sqlite3.DatabaseError): + self.model() + # An invalid annotation group invalidates the review group beside it. + path = self.tampered( + "bad annotations", + ("UPDATE media_annotation_bytes SET sha256=upper(sha256)",), + checks_off=True, + ) + with closing(sqlite3.connect(path)) as raw: + self.assertIn("image annotation", review_schema_error(raw).lower()) + + def tampered(self, name, statements, *, checks_off=False, rows_in=True): + """A closed current database whose group was altered after the fact.""" + path = str(Path(self.temp.name) / f"{name.replace(' ', '-')}.db") + model = self.model(path) + with model.get_connection() as conn: + populate_legacy(conn) + if rows_in: + insert_group_rows(conn) + model.close_all() + with closing(sqlite3.connect(path)) as raw, raw: + raw.execute("PRAGMA foreign_keys=OFF") + if checks_off: + raw.execute("PRAGMA ignore_check_constraints=ON") + for statement in statements: + if isinstance(statement, tuple): + raw.execute(*statement) + else: + raw.execute(statement) + return path + + def assert_rejected_unrepaired(self, path): + with closing(sqlite3.connect(path)) as raw: + schema_before = raw.execute( + "SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY name" + ).fetchall() + rows_before = table_rows(raw, TABLES) + for allow_missing in (False, True): + error = review_schema_error(raw, allow_missing=allow_missing) + self.assertIsNotNone(error) + self.assertIn("review", error.lower()) + self.assertFalse(PromptModel.verify_database_file(path)[0]) + self.assertFalse( + PromptModel.verify_database_file(path, require_library_state=True)[0] + ) + PromptModel.reset_schema_cache() + with self.assertRaises(sqlite3.DatabaseError): + self.model(path) + PromptModel.reset_schema_cache() + with closing(sqlite3.connect(path)) as raw: + self.assertEqual( + raw.execute( + "SELECT type,name,tbl_name,sql FROM sqlite_master ORDER BY name" + ).fetchall(), + schema_before, + ) + self.assertEqual(table_rows(raw, TABLES), rows_before) + + def test_partial_or_malformed_group_is_rejected_without_repair(self): + self.require_module() + cases = { + "missing metadata": (f"DROP TABLE {METADATA}",), + "missing items": (f"DROP TABLE {DECISIONS}", f"DROP TABLE {ITEMS}"), + "missing decisions": (f"DROP TABLE {DECISIONS}",), + "reserved view": ( + f"DROP TABLE {DECISIONS}", + f"CREATE VIEW {DECISIONS} AS SELECT 'x' AS id", + ), + "extra trigger": ( + f"CREATE TRIGGER media_review_trigger AFTER INSERT ON {ITEMS} " + "BEGIN SELECT 1; END", + ), + "extra index": ( + f"CREATE INDEX idx_media_review_extra ON {ITEMS}(updated_at)", + ), + "extra column": ( + f"DROP TABLE {DECISIONS}", + DDL[DECISIONS].replace("evidence TEXT", "note TEXT, evidence TEXT"), + ), + "wrong primary key": ( + f"DROP TABLE {DECISIONS}", + DDL[DECISIONS].replace("id TEXT PRIMARY KEY", "id TEXT", 1), + ), + "wrong foreign key": ( + f"DROP TABLE {DECISIONS}", + DDL[DECISIONS].replace( + "generated_images(id) ON DELETE SET NULL", + "generated_images(id) ON DELETE CASCADE", + ), + INDEXES["idx_media_review_decisions_pair"], + INDEXES["idx_media_review_decisions_image"], + ), + "wrong check": ( + f"DROP TABLE {METADATA}", + DDL[METADATA].replace("schema_version=1", "schema_version=2"), + f"INSERT INTO {METADATA} VALUES(1,2,0)", + ), + "missing unique": ( + "DROP INDEX idx_media_review_decisions_pair", + "DROP INDEX idx_media_review_decisions_image", + f"DROP TABLE {DECISIONS}", + DDL[DECISIONS].replace(" UNIQUE(item_id, decision_revision),", ""), + INDEXES["idx_media_review_decisions_pair"], + INDEXES["idx_media_review_decisions_image"], + ), + "missing queue index": ("DROP INDEX idx_media_review_items_queue",), + "wrong index columns": ( + "DROP INDEX idx_media_review_items_queue", + INDEXES["idx_media_review_items_queue"].replace(", id", ""), + ), + "orphan item": ((f"UPDATE {ITEMS} SET byte_id=?", (OTHER_ID,)),), + "orphan decision": ((f"UPDATE {DECISIONS} SET item_id=?", (OTHER_ID,)),), + "dangling pair": (f"UPDATE {DECISIONS} SET pair_id=404",), + "bad item uuid": ( + (f"UPDATE {ITEMS} SET id=?", ("x" * 36,)), + (f"UPDATE {DECISIONS} SET item_id=?", ("x" * 36,)), + ), + "uppercase decision uuid": (f"UPDATE {DECISIONS} SET id=upper(id)",), + "null decision id": (f"UPDATE {DECISIONS} SET id=NULL",), + "blob ids": (f"UPDATE {DECISIONS} SET id=CAST(id AS BLOB)",), + "no metadata row": (f"DELETE FROM {METADATA}",), + "long created_at": ((f"UPDATE {ITEMS} SET created_at=?", ("x" * 65,)),), + "blob updated_at": (f"UPDATE {ITEMS} SET updated_at=X'00'",), + "text resolver version": (f"UPDATE {ITEMS} SET resolver_version='one'",), + "evidence hash mismatch": ( + (f"UPDATE {ITEMS} SET evidence_hash=?", (CANDIDATE,)), + ), + } + for label, statements in cases.items(): + with self.subTest(label): + self.assert_rejected_unrepaired(self.tampered(label, statements)) + unchecked = { + "two metadata rows": (f"INSERT INTO {METADATA} VALUES(2,1,0)",), + "wrong schema version": (f"UPDATE {METADATA} SET schema_version=2",), + "negative index revision": ( + f"UPDATE {METADATA} SET review_index_revision=-1", + ), + "uppercase evidence hash": ( + f"UPDATE {ITEMS} SET evidence_hash=upper(evidence_hash)", + ), + "bad state": (f"UPDATE {ITEMS} SET state='done'",), + "limited two": (f"UPDATE {ITEMS} SET limited=2",), + "negative revision": (f"UPDATE {ITEMS} SET revision=-1",), + "negative candidates": (f"UPDATE {ITEMS} SET candidate_count=-1",), + "bad outcome": (f"UPDATE {DECISIONS} SET outcome='maybe'",), + "short candidate hash": ( + f"UPDATE {DECISIONS} SET candidate_hash=substr(candidate_hash,1,63)", + ), + "negative decision revision": ( + f"UPDATE {DECISIONS} SET decision_revision=-1", + ), + } + for label, statements in unchecked.items(): + with self.subTest(label): + path = self.tampered(label, statements, checks_off=True) + self.assert_rejected_unrepaired(path) + + def test_selected_decision_requires_candidate_hash(self): + self.require_module() + model = self.model() + with model.get_connection() as conn: + insert_byte(conn) + insert_item(conn) + for outcome, candidate in (("selected", None), ("unresolved", CANDIDATE)): + with self.subTest(outcome): + with self.assertRaises(sqlite3.IntegrityError): + insert_decision(conn, outcome=outcome, candidate_hash=candidate) + insert_decision(conn, outcome="unresolved", candidate_hash=None) + insert_decision(conn, decision_id=OTHER_ID, decision_revision=2) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + self.assertEqual( + rows(conn, f"SELECT outcome FROM {DECISIONS} ORDER BY 1"), + [("selected",), ("unresolved",)], + ) + with self.assertRaises(sqlite3.IntegrityError): # one per revision + insert_decision(conn, decision_id=ITEM_ID, decision_revision=2) + model.close_all() + path = self.tampered( + "selected without hash", + (f"UPDATE {DECISIONS} SET candidate_hash=NULL",), + checks_off=True, + ) + self.assert_rejected_unrepaired(path) + + def test_oversize_or_noncanonical_evidence_rejected(self): + self.require_module() + self.assertEqual( + canonical_evidence({"b": [1, "é"], "a": None}), + '{"a":null,"b":[1,"\\u00e9"]}', + ) + self.assertEqual( + evidence_digest(EVIDENCE), hashlib.sha256(EVIDENCE.encode()).hexdigest() + ) + for bad in ( + [], + "text", + {"x": float("nan")}, + {"x": float("inf")}, + {"x": object()}, + {1: 1, "a": 2}, + ): + with self.subTest(value=repr(bad)), self.assertRaises(ValueError): + canonical_evidence(bad) + with self.assertRaises(ValueError): + canonical_evidence({"pad": "x" * MAX_EVIDENCE_BYTES}) + oversize = '{"pad":"' + "x" * (MAX_EVIDENCE_BYTES - 9) + '"}' + self.assertEqual(len(oversize), MAX_EVIDENCE_BYTES + 1) + model = self.model() + with model.get_connection() as conn: + insert_byte(conn) + with self.assertRaises(sqlite3.IntegrityError): + insert_item(conn, evidence=oversize) + model.close_all() + noncanonical = { + "spaces": '{"candidates": [], "sha256": "x"}', + "unsorted": '{"sha256":"x","candidates":[]}', + "raw unicode": '{"text":"é"}', + "duplicate key": '{"a":1,"a":2}', + "not an object": "[]", + "not json": "{", + "nan": '{"a":NaN}', + "empty": "", + } + for label, evidence in noncanonical.items(): + for table in (ITEMS, DECISIONS): + with self.subTest(label, table=table): + digest = hashlib.sha256(evidence.encode("utf-8")).hexdigest() + statements = [ + (f"UPDATE {table} SET evidence=?", (evidence,)), + (f"UPDATE {ITEMS} SET evidence_hash=?", (digest,)), + ] + if table == DECISIONS: + statements.pop() + self.assert_rejected_unrepaired( + self.tampered(f"{label} {table}", statements) + ) + for table in (ITEMS, DECISIONS): + with self.subTest("oversize", table=table): + digest = hashlib.sha256(oversize.encode("ascii")).hexdigest() + statements = [ + (f"UPDATE {table} SET evidence=?", (oversize,)), + (f"UPDATE {ITEMS} SET evidence_hash=?", (digest,)), + ] + path = self.tampered(f"oversize {table}", statements, checks_off=True) + self.assert_rejected_unrepaired(path) + + def test_seventeen_candidates_rejected(self): + self.require_module() + model = self.model() + with model.get_connection() as conn: + insert_byte(conn) + with self.assertRaises(sqlite3.IntegrityError): + insert_item(conn, candidates=17) + insert_item(conn, candidates=16) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + model.close_all() + path = self.tampered( + "seventeen", (f"UPDATE {ITEMS} SET candidate_count=17",), checks_off=True + ) + self.assert_rejected_unrepaired(path) + + def test_prompt_or_image_deletion_nulls_navigation_and_keeps_decision(self): + self.require_module() + db = PromptDatabase(self.path) + self.models.append(db.model) + with db.model.get_connection() as conn: + populate_legacy(conn) + conn.execute( + "INSERT INTO prompt_pairs(id,positive_prompt_id,negative_prompt_id) " + "VALUES(8,9,NULL)" + ) + insert_byte(conn) + insert_item(conn) + insert_decision(conn, pair_id=8, image_id=4) + conn.commit() + self.assertTrue(db.delete_prompt_pair(8)) + with db.model.get_connection() as conn: + self.assertEqual( + rows(conn, f"SELECT pair_id, image_id FROM {DECISIONS}"), [(None, 4)] + ) + with db.model.get_connection() as conn: + conn.execute("DELETE FROM prompt_pairs WHERE id=6") + conn.commit() + self.assertTrue(db.delete_prompt(7)) + with db.model.get_connection() as conn: + self.assertEqual( + rows( + conn, + f"SELECT id, item_id, outcome, candidate_hash, pair_id, image_id " + f"FROM {DECISIONS}", + ), + [(DECISION_ID, ITEM_ID, "selected", CANDIDATE, None, None)], + ) + self.assertEqual(rows(conn, f"SELECT id FROM {ITEMS}"), [(ITEM_ID,)]) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + # The SET NULL actions look decisions up by index, not by a scan. + for column in ("pair_id", "image_id"): + plan = " ".join( + str(row[3]) + for row in conn.execute( + f"EXPLAIN QUERY PLAN UPDATE {DECISIONS} " # nosec B608 + f"SET {column}=NULL WHERE {column}=?", + (1,), + ) + ) + self.assertIn(f"idx_media_review_decisions_{column[:-3]}", plan) + + def test_full_capacity_unicode_control_pair_round_trips_within_8_mib(self): + self.require_module() + value = full_capacity_evidence() + evidence = canonical_evidence(value) + self.assertEqual(len(evidence.encode("utf-8")), MAX_EVIDENCE_BYTES) + self.assertTrue(evidence.isascii()) + source_path = str(Path(self.temp.name) / "source.db") + source = self.model(source_path) + with source.get_connection() as conn: + conn.execute("PRAGMA wal_autocheckpoint=0") + insert_byte(conn) + insert_item(conn, evidence=evidence, candidates=16) + insert_decision(conn, evidence=evidence) + conn.commit() + source_rows = table_rows(conn, TABLES) + self.assertGreater(Path(source_path + "-wal").stat().st_size, 0) + live = self.model() + backup_file = str(Path(self.temp.name) / "backup.db") + self.assertTrue(source.backup_database(backup_file)) + for restore_from in (source_path, backup_file): + with self.subTest(Path(restore_from).name): + live.restore_from_file(restore_from) + with live.get_connection() as conn: + self.assertEqual(table_rows(conn, TABLES), source_rows) + self.assertIsNone(review_schema_error(conn, allow_missing=False)) + stored = conn.execute( + f"SELECT i.evidence, d.evidence FROM {ITEMS} i " + f"JOIN {DECISIONS} d ON d.item_id = i.id" + ).fetchone() + self.assertEqual(json.loads(stored[0]), value) + self.assertEqual(json.loads(stored[1]), value) + PromptModel.close_all_instances() + assert_closed(self, self.temp.name) + + def test_install_failure_rolls_back_and_retries(self): + self.require_module() + with patch( + "database.review_schema.review_schema_error", + side_effect=[None, "injected final validation failure"], + ): + with self.assertRaisesRegex(sqlite3.DatabaseError, "injected"): + self.model() + with closing(sqlite3.connect(self.path)) as raw: + self.assertEqual(catalog(raw), set()) + self.assertIsNotNone( + raw.execute( + "SELECT 1 FROM sqlite_master WHERE name='media_annotation_bytes'" + ).fetchone() + ) + PromptModel.reset_schema_cache() + model = self.model() + self.assertEqual( + PromptModel.verify_database_file(model.db_path, require_library_state=True), + (True, "ok"), + ) + with model.get_connection() as conn: + self.assertEqual(metadata(conn), [(1, 0)]) + + def test_wal_backup_restore_round_trips_group(self): + self.require_module() + source_path = str(Path(self.temp.name) / "source.db") + source = self.model(source_path) + with source.get_connection() as conn: + conn.execute("PRAGMA wal_autocheckpoint=0") + insert_group_rows(conn) + conn.execute(f"UPDATE {METADATA} SET review_index_revision=3") + conn.commit() + source_rows = table_rows(conn, TABLES) + self.assertGreater(Path(source_path + "-wal").stat().st_size, 0) + live = self.model() + with live.get_connection() as conn: + insert_byte(conn) + insert_item(conn, item_id=OTHER_ID) + source_before = Path(source_path).read_bytes() + backup_file = str(Path(self.temp.name) / "backup.db") + self.assertTrue(source.backup_database(backup_file)) + with closing(sqlite3.connect(backup_file)) as raw: + self.assertEqual(table_rows(raw, TABLES), source_rows) + self.assertIsNone(review_schema_error(raw, allow_missing=False)) + + safety = live.restore_from_file(source_path) + + self.assertEqual(Path(source_path).read_bytes(), source_before) + with live.get_connection() as conn: + self.assertEqual(table_rows(conn, TABLES), source_rows) + self.assertEqual(metadata(conn), [(1, 3)]) + with closing(sqlite3.connect(safety)) as raw: + self.assertEqual(rows(raw, f"SELECT id FROM {ITEMS}"), [(OTHER_ID,)]) + PromptModel.reset_schema_cache() + live.restore_from_file(backup_file) + with live.get_connection() as conn: + self.assertEqual(table_rows(conn, TABLES), source_rows) + PromptModel.close_all_instances() + assert_closed(self, self.temp.name) + + def test_staged_failure_preserves_source_live_safety_and_handles(self): + self.require_module() + live = self.model() + with live.get_connection() as conn: + insert_byte(conn) + insert_item(conn, item_id=OTHER_ID) + live_rows = table_rows(conn, TABLES) + bad = self.tampered( + "restore bad", (f"UPDATE {ITEMS} SET state='done'",), checks_off=True + ) + old = str(Path(self.temp.name) / "old.db") + self.model(old).close_all() + self.drop_tables(old) + for source, patched in ((bad, False), (old, True)): + with self.subTest(source=Path(source).name): + before = Path(source).read_bytes() + safety = sorted(Path(self.temp.name).glob("live.db.backup_*")) + if patched: + context = patch( + "database.models.migrate_review_state", + side_effect=sqlite3.OperationalError("staging failed"), + ) + expected = sqlite3.OperationalError + else: + context = patch.object(PromptModel, "backup_database") + expected = ValueError + with context as hook, self.assertRaises(expected): + live.restore_from_file(source) + if not patched: + hook.assert_not_called() + self.assertEqual(Path(source).read_bytes(), before) + self.assertEqual( + sorted(Path(self.temp.name).glob("live.db.backup_*")), safety + ) + assert_closed(self, source) + with live.get_connection() as conn: + self.assertEqual(table_rows(conn, TABLES), live_rows) + self.assertFalse( + PromptModel.verify_database_file(old, require_library_state=True)[0] + ) + self.assertTrue(PromptModel.verify_database_file(old)[0]) + + def test_verify_requires_group_only_with_require_library_state(self): + self.require_module() + self.model().close_all() + self.assertEqual( + PromptModel.verify_database_file(self.path, require_library_state=True), + (True, "ok"), + ) + self.drop_tables(self.path) + self.assertEqual(PromptModel.verify_database_file(self.path), (True, "ok")) + self.assertEqual( + PromptModel.verify_database_file(self.path, require_library_state=True), + (False, "Missing review queue schema"), + ) + + +if __name__ == "__main__": + unittest.main()