From daa41ff6cca9e5bfa3df801c658dee935fac3790 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Wed, 21 Oct 2026 04:34:44 -0700 Subject: [PATCH] feat(views): saved-views store and API Add a saved_views table (page images|library, trimmed 1-100 character name, whitelisted JSON state of at most 4 KB, position, timestamps) installed through the same validated-savepoint migration as image favorites. It is optional in old backups and required in a migrated restore stage; restore rejects a tampered definition, extra indexes and any row that would fail write validation. Routes under /prompt_manager/saved-views: GET lists views by position, POST {page,name,state} returns 201 {success, view}, PATCH /{id} {name} renames and DELETE /{id} removes, both 404 for an unknown id. The state must hold exactly the page's toolbar and Refine-this-view keys (Images: q, category, sort, source, folder, media, tags; Library: q, category, sort, tags) with their types and sort/media values; anything else is a 400. Refs #249 --- database/models.py | 12 +- database/operations.py | 2 + database/saved_views.py | 248 +++++++++++++++++++++++++ py/api/__init__.py | 3 + py/api/saved_views.py | 107 +++++++++++ tests/test_saved_views.py | 377 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 747 insertions(+), 2 deletions(-) create mode 100644 database/saved_views.py create mode 100644 py/api/saved_views.py create mode 100644 tests/test_saved_views.py diff --git a/database/models.py b/database/models.py index 85f1f23..8e9e82f 100644 --- a/database/models.py +++ b/database/models.py @@ -17,6 +17,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 .saved_views import migrate_saved_views, saved_views_schema_error from .execution_history_schema import ( history_schema_error, migrate_execution_history, @@ -300,6 +301,7 @@ class PromptModel: migrate_execution_history(conn) migrate_library_state(conn) migrate_image_favorites(conn) + migrate_saved_views(conn) def _migrate_prompt_pairs(self, conn: sqlite3.Connection) -> None: conn.execute("SAVEPOINT prompt_pair_schema") @@ -1019,8 +1021,9 @@ class PromptModel: Args: path: File to inspect require_library_state: Also require the tables every current - database has (Library state and image favorites), as a - migrated restore stage must; old backups may lack them. + database has (Library state, image favorites and saved + views), as a migrated restore stage must; old backups may + lack them. Returns: (ok, reason): reason is "ok" when the file is acceptable, otherwise @@ -1080,6 +1083,11 @@ class PromptModel: ) if favorites_error: return False, favorites_error + views_error = saved_views_schema_error( + conn, allow_missing=not require_library_state + ) + if views_error: + return False, views_error return True, "ok" except sqlite3.Error as e: return False, f"Invalid SQLite database: {e}" diff --git a/database/operations.py b/database/operations.py index 6abdb05..2584498 100644 --- a/database/operations.py +++ b/database/operations.py @@ -12,6 +12,7 @@ from .models import PromptModel, normalize_image_path, utc_now_iso from .pairs import PairDependencyError, PairOperations from .library import LibraryOperations, titled from .image_favorites import ImageFavoriteOperations +from .saved_views import SavedViewOperations from .library_commands import LibraryCommandOperations from .vocabulary import VocabularyOperations from .execution_history import ExecutionHistoryOperations @@ -306,6 +307,7 @@ class PromptDatabase( ExecutionOutputOperations, HistoryReadOperations, ImageFavoriteOperations, + SavedViewOperations, ): """Database operations class for managing prompts.""" diff --git a/database/saved_views.py b/database/saved_views.py new file mode 100644 index 0000000..3c3fc96 --- /dev/null +++ b/database/saved_views.py @@ -0,0 +1,248 @@ +"""Persisted saved views (D16): a named Images or Library toolbar state. + +A view's state holds exactly its page's toolbar values and its Refine-this-view +refinements, whitelisted per page and stored as compact JSON of at most 4 KB. +The same validation guards writes and restores. Optional in old backups, +required once a database is current (see ``library_schema.py``). +""" + +import json +import re +import sqlite3 +from typing import Optional + +TABLE = "saved_views" +PAGES = ("images", "library") +# The schema check's literal SQL repeats the name and state caps (names in +# bytes, at most 4 per character); keep them in step. +MAX_NAME_LENGTH = 100 +MAX_STATE_BYTES = 4096 +MAX_TAGS = 20 +MAX_TAG_LENGTH = 50 +_IMAGE_SORTS = ("newest", "oldest", "name", "name-desc", "rating") +_LIBRARY_SORTS = ("newest", "oldest", "name", "rating") +_MEDIA = ("all", "image", "video") +_CONTROL = re.compile(r"[\x00-\x1f\x7f]") +_SOURCE = re.compile(r"all|0|[1-9][0-9]{0,8}") + +# (kind, limit-or-choices) per state key; "text" values are length-capped. +_STATE_FIELDS = { + "images": { + "q": ("text", 500), + "category": ("text", 100), + "sort": ("choice", _IMAGE_SORTS), + "source": ("source", None), + "folder": ("text", 1024), + "media": ("choice", _MEDIA), + "tags": ("tags", None), + }, + "library": { + "q": ("text", 500), + "category": ("text", 100), + "sort": ("choice", _LIBRARY_SORTS), + "tags": ("tags", None), + }, +} + +_DDL = ( + f"CREATE TABLE {TABLE} (id INTEGER PRIMARY KEY AUTOINCREMENT, " + "page TEXT NOT NULL CHECK(page IN ('images','library')), " + "name TEXT NOT NULL, state TEXT NOT NULL, position INTEGER NOT NULL, " + "created_at TEXT NOT NULL, updated_at TEXT NOT NULL)" +) +_COLUMNS = ( + ("id", "INTEGER", 0, 1), + ("page", "TEXT", 1, 0), + ("name", "TEXT", 1, 0), + ("state", "TEXT", 1, 0), + ("position", "INTEGER", 1, 0), + ("created_at", "TEXT", 1, 0), + ("updated_at", "TEXT", 1, 0), +) +_SELECT = "SELECT id, page, name, state FROM saved_views" + + +def _compact(sql: str) -> str: + return re.sub(r"\s+", "", sql).lower() + + +def validate_name(name: object) -> str: + """The trimmed name, 1 to 100 characters without control characters.""" + if not isinstance(name, str): + raise ValueError("Name must be a string") + trimmed = name.strip() + if not 1 <= len(trimmed) <= MAX_NAME_LENGTH: + raise ValueError(f"Name must be 1 to {MAX_NAME_LENGTH} characters") + if _CONTROL.search(trimmed): + raise ValueError("Name contains control characters") + return trimmed + + +def _validate_value(key: str, kind: str, limit, value: object) -> None: + if kind == "tags": + if not isinstance(value, list) or len(value) > MAX_TAGS: + raise ValueError(f"State '{key}' must be a list of at most {MAX_TAGS}") + for tag in value: + if ( + not isinstance(tag, str) + or not tag.strip() + or len(tag) > MAX_TAG_LENGTH + or "," in tag + or _CONTROL.search(tag) + ): + raise ValueError(f"State '{key}' holds an invalid tag") + return + if not isinstance(value, str): + raise ValueError(f"State '{key}' must be a string") + if kind == "choice" and value not in limit: + raise ValueError(f"State '{key}' is not a supported value") + if kind == "source" and not _SOURCE.fullmatch(value): + raise ValueError(f"State '{key}' must be 'all' or a source index") + if kind == "text" and (len(value) > limit or _CONTROL.search(value)): + raise ValueError(f"State '{key}' is too long or has control characters") + + +def encode_state(page: object, state: object) -> str: + """Validate *state* for *page* and return its compact JSON (at most 4 KB).""" + if page not in PAGES: + raise ValueError("Page must be 'images' or 'library'") + fields = _STATE_FIELDS[page] + if not isinstance(state, dict): + raise ValueError("State must be an object") + if set(state) != set(fields): + raise ValueError( + f"State for {page} must contain exactly " + ", ".join(sorted(fields)) + ) + for key, (kind, limit) in fields.items(): + _validate_value(key, kind, limit, state[key]) + encoded = json.dumps( + {key: state[key] for key in fields}, ensure_ascii=False, separators=(",", ":") + ) + if len(encoded.encode("utf-8")) > MAX_STATE_BYTES: + raise ValueError("State must encode to at most 4 KB") + return encoded + + +def _row_error(page, name, state) -> bool: + """True when a persisted row would not pass the write validation.""" + try: + if validate_name(name) != name: + return True + decoded = json.loads(state) + return encode_state(page, decoded) != state + except (ValueError, TypeError): + return True + + +def saved_views_schema_error( + conn: sqlite3.Connection, *, allow_missing: bool = True +) -> Optional[str]: + """Reject unsupported saved-views schemas and invalid persisted values. + + Reads only. A missing table is acceptable for backups that predate it + unless ``allow_missing`` is false. + """ + objects = conn.execute( + "SELECT type,name,sql FROM sqlite_master WHERE name=? " + "OR (type IN ('index','trigger') AND tbl_name=?)", + (TABLE, TABLE), + ).fetchall() + if not objects: + return None if allow_missing else "Missing saved views schema" + tables = [sql for kind, _name, sql in objects if kind == "table"] + if len(tables) != 1 or len(objects) != 1: + return "Saved views name is reserved by an unsupported schema object" + if _compact(tables[0] or "") != _compact(_DDL): + return "Saved views has an invalid table definition" + columns = tuple( + (row[1], row[2].upper(), row[3], row[5]) + for row in conn.execute(f"PRAGMA table_info({TABLE})") + ) + if columns != _COLUMNS: + return "Saved views has invalid columns" + # Literal SQL: the table name and limits are constants of this module. + if conn.execute( + "SELECT 1 FROM saved_views WHERE typeof(page) != 'text' " + "OR typeof(name) != 'text' OR typeof(state) != 'text' " + "OR typeof(position) != 'integer' OR typeof(created_at) != 'text' " + "OR typeof(updated_at) != 'text' " + "OR length(CAST(name AS BLOB)) > 400 " + "OR length(CAST(state AS BLOB)) > 4096 " + "OR length(CAST(created_at AS BLOB)) > 64 " + "OR length(CAST(updated_at AS BLOB)) > 64 LIMIT 1" + ).fetchone(): + return "Saved views contains invalid values" + for page, name, state in conn.execute("SELECT page, name, state FROM saved_views"): + if _row_error(page, name, state): + return "Saved views contains invalid values" + return None + + +def migrate_saved_views(conn: sqlite3.Connection) -> None: + """Install the table atomically; validate before and after creating it.""" + conn.execute("SAVEPOINT saved_views_schema") + try: + error = saved_views_schema_error(conn) + if error: + raise sqlite3.DatabaseError(error) + conn.execute(_DDL.replace("CREATE TABLE", "CREATE TABLE IF NOT EXISTS", 1)) + error = saved_views_schema_error(conn, allow_missing=False) + if error: + raise sqlite3.DatabaseError(error) + conn.execute("RELEASE saved_views_schema") + except BaseException: + conn.execute("ROLLBACK TO saved_views_schema") + conn.execute("RELEASE saved_views_schema") + raise + + +def _view(row) -> dict: + view_id, page, name, state = row + return {"id": view_id, "page": page, "name": name, "state": json.loads(state)} + + +class SavedViewOperations: + """PromptDatabase mixin: list, create, rename and delete saved views.""" + + def list_saved_views(self) -> list[dict]: + """Every view, in list order (position, then creation).""" + rows = self.model.get_connection().execute(_SELECT + " ORDER BY position, id") + return [_view(row) for row in rows] + + def create_saved_view(self, page: object, name: object, state: object) -> dict: + """Validate and append one view at the end of the list.""" + from .models import utc_now_iso # models imports this module's schema + + encoded = encode_state(page, state) + trimmed = validate_name(name) + now = utc_now_iso() + with self.model.get_connection() as conn: + cursor = conn.execute( + "INSERT INTO saved_views (page, name, state, position, " + "created_at, updated_at) SELECT ?, ?, ?, " + "COALESCE(MAX(position), -1) + 1, ?, ? FROM saved_views", + (page, trimmed, encoded, now, now), + ) + row = conn.execute( + _SELECT + " WHERE id = ?", (cursor.lastrowid,) + ).fetchone() + return _view(row) + + def rename_saved_view(self, view_id: int, name: object) -> Optional[dict]: + """The renamed view, or None when no view has *view_id*.""" + from .models import utc_now_iso + + trimmed = validate_name(name) + with self.model.get_connection() as conn: + conn.execute( + "UPDATE saved_views SET name = ?, updated_at = ? WHERE id = ?", + (trimmed, utc_now_iso(), view_id), + ) + row = conn.execute(_SELECT + " WHERE id = ?", (view_id,)).fetchone() + return _view(row) if row else None + + def delete_saved_view(self, view_id: int) -> bool: + """True when a view was removed.""" + with self.model.get_connection() as conn: + cursor = conn.execute("DELETE FROM saved_views WHERE id = ?", (view_id,)) + return cursor.rowcount > 0 diff --git a/py/api/__init__.py b/py/api/__init__.py index 2e15b90..e9649d6 100644 --- a/py/api/__init__.py +++ b/py/api/__init__.py @@ -25,6 +25,7 @@ from .prompts import PromptRoutesMixin from .pairs import PairRoutesMixin from .library import LibraryRoutesMixin from .vocabulary import VocabularyRoutesMixin +from .saved_views import SavedViewRoutesMixin from .history import HistoryRoutesMixin from .images import ImageRoutesMixin from .admin import AdminRoutesMixin @@ -248,6 +249,7 @@ class PromptManagerAPI( PairRoutesMixin, LibraryRoutesMixin, VocabularyRoutesMixin, + SavedViewRoutesMixin, HistoryRoutesMixin, PromptRoutesMixin, ImageRoutesMixin, @@ -432,6 +434,7 @@ class PromptManagerAPI( self._register_pair_routes(routes) self._register_library_routes(routes) self._register_vocabulary_routes(routes) + self._register_saved_view_routes(routes) self._register_history_routes(routes) self._register_prompt_routes(routes) self._register_image_routes(routes) diff --git a/py/api/saved_views.py b/py/api/saved_views.py new file mode 100644 index 0000000..5000fd5 --- /dev/null +++ b/py/api/saved_views.py @@ -0,0 +1,107 @@ +"""Saved views (D16): list, create, rename and delete named toolbar states. + +The page, the name and the whitelisted per-page state are validated here +(through ``database/saved_views.py``) before anything is written. +""" + +from functools import wraps +from typing import Callable + +from aiohttp import web + +try: + from ...database.saved_views import encode_state, validate_name +except ImportError: + from database.saved_views import encode_state, validate_name + +_CREATE_FIELDS = frozenset({"page", "name", "state"}) +_RENAME_FIELDS = frozenset({"name"}) +_MAX_ID = 2**63 - 1 + + +class SavedViewNotFound(LookupError): + pass + + +def saved_view_response(handler: Callable) -> Callable: + """``{success:false,error}`` with 400, 404 or 500 on every failure.""" + + @wraps(handler) + async def wrapped(self, request: web.Request) -> web.Response: + try: + return await handler(self, request) + except SavedViewNotFound: + return web.json_response( + {"success": False, "error": "Saved view not found"}, status=404 + ) + except (ValueError, TypeError) as exc: + return web.json_response({"success": False, "error": str(exc)}, status=400) + except Exception: + self.logger.exception("Saved view operation failed") + return web.json_response( + {"success": False, "error": "Saved view operation failed"}, status=500 + ) + + return wrapped + + +async def _body(request: web.Request, fields: frozenset) -> dict: + """A JSON object whose keys are exactly ``fields``.""" + body = await request.json() + if not isinstance(body, dict): + raise ValueError("Expected a JSON object") + if set(body) != fields: + raise ValueError("Request must contain exactly " + ", ".join(sorted(fields))) + return body + + +def _view_id(request: web.Request) -> int: + raw = request.match_info["view_id"] + if not raw.isascii() or not raw.isdigit(): + raise ValueError("Invalid saved view ID") + value = int(raw) + if not 1 <= value <= _MAX_ID: + raise ValueError("Invalid saved view ID") + return value + + +class SavedViewRoutesMixin: + def _register_saved_view_routes(self, routes: web.RouteTableDef) -> None: + routes.get("/prompt_manager/saved-views")(self.list_saved_views) + routes.post("/prompt_manager/saved-views")(self.create_saved_view) + routes.patch("/prompt_manager/saved-views/{view_id}")(self.rename_saved_view) + routes.delete("/prompt_manager/saved-views/{view_id}")(self.delete_saved_view) + + @saved_view_response + async def list_saved_views(self, request: web.Request) -> web.Response: + views = await self._run_in_executor(self.db.list_saved_views) + return web.json_response({"success": True, "views": views}) + + @saved_view_response + async def create_saved_view(self, request: web.Request) -> web.Response: + body = await _body(request, _CREATE_FIELDS) + encode_state(body["page"], body["state"]) + validate_name(body["name"]) + view = await self._run_in_executor( + self.db.create_saved_view, body["page"], body["name"], body["state"] + ) + return web.json_response({"success": True, "view": view}, status=201) + + @saved_view_response + async def rename_saved_view(self, request: web.Request) -> web.Response: + view_id = _view_id(request) + body = await _body(request, _RENAME_FIELDS) + validate_name(body["name"]) + view = await self._run_in_executor( + self.db.rename_saved_view, view_id, body["name"] + ) + if view is None: + raise SavedViewNotFound + return web.json_response({"success": True, "view": view}) + + @saved_view_response + async def delete_saved_view(self, request: web.Request) -> web.Response: + view_id = _view_id(request) + if not await self._run_in_executor(self.db.delete_saved_view, view_id): + raise SavedViewNotFound + return web.json_response({"success": True, "id": view_id}) diff --git a/tests/test_saved_views.py b/tests/test_saved_views.py new file mode 100644 index 0000000..568c276 --- /dev/null +++ b/tests/test_saved_views.py @@ -0,0 +1,377 @@ +"""Saved views (D16): the persisted store, its restore schema and the API.""" + +import json +import os +import sqlite3 +import tempfile +import unittest +from contextlib import closing +from pathlib import Path + +try: + from test_api_images import ImageAPITestCase + from open_handles import assert_closed +except ImportError: + from tests.test_api_images import ImageAPITestCase + from tests.open_handles import assert_closed + +from database.models import PromptModel +from database.operations import PromptDatabase +from database.saved_views import ( + MAX_STATE_BYTES, + TABLE, + migrate_saved_views, + saved_views_schema_error, +) +from utils.hashing import generate_prompt_hash + +URL = "/prompt_manager/saved-views" + +IMAGES_STATE = { + "q": "cat", + "category": "Animals", + "sort": "newest", + "source": "all", + "folder": "", + "media": "all", + "tags": ["fluffy"], +} +LIBRARY_STATE = {"q": "", "category": "", "sort": "rating", "tags": []} + + +class TestSavedViewsApi(ImageAPITestCase): + async def send(self, method, path="", body=None, **kwargs): + if body is not None: + kwargs["json"] = body + response = await self.client.request(method, URL + path, **kwargs) + return response, await response.json() + + async def create(self, page="images", name="Cats", state=None): + state = dict( + IMAGES_STATE if page == "images" else LIBRARY_STATE, **(state or {}) + ) + response, data = await self.send( + "POST", body={"page": page, "name": name, "state": state} + ) + self.assertEqual(response.status, 201, data) + return data["view"] + + async def test_create_list_rename_delete_round_trip(self): + response, data = await self.send("GET") + self.assertEqual(response.status, 200) + self.assertEqual(data, {"success": True, "views": []}) + + first = await self.create(name=" Cats ") + self.assertEqual( + first, + { + "id": first["id"], + "page": "images", + "name": "Cats", + "state": IMAGES_STATE, + }, + ) + second = await self.create(page="library", name="Best", state={}) + _, data = await self.send("GET") + self.assertTrue(data["success"]) + self.assertEqual(data["views"], [first, second]) + + response, data = await self.send( + "PATCH", f"/{first['id']}", {"name": " Kittens "} + ) + self.assertEqual(response.status, 200) + self.assertEqual(data, {"success": True, "view": {**first, "name": "Kittens"}}) + _, listed = await self.send("GET") + # Renaming keeps the view's place in the list. + self.assertEqual( + [view["name"] for view in listed["views"]], ["Kittens", "Best"] + ) + + response, data = await self.send("DELETE", f"/{first['id']}") + self.assertEqual(response.status, 200) + self.assertEqual(data, {"success": True, "id": first["id"]}) + _, listed = await self.send("GET") + self.assertEqual(listed["views"], [second]) + + for method, body in (("PATCH", {"name": "Again"}), ("DELETE", None)): + response, data = await self.send(method, f"/{first['id']}", body) + self.assertEqual(response.status, 404) + self.assertFalse(data["success"]) + self.assertIn("error", data) + + async def test_views_are_ordered_by_position_and_ids_are_not_reused(self): + views = [await self.create(name=f"View {n}") for n in range(3)] + await self.send("DELETE", f"/{views[2]['id']}") + later = await self.create(name="Later") + self.assertGreater(later["id"], views[2]["id"]) + with self.api.db.model.get_connection() as conn: + conn.execute( + f"UPDATE {TABLE} SET position = -1 WHERE id = ?", (views[1]["id"],) + ) + _, listed = await self.send("GET") + self.assertEqual( + [view["name"] for view in listed["views"]], ["View 1", "View 0", "Later"] + ) + + async def test_create_rejects_invalid_bodies(self): + good = {"page": "images", "name": "Cats", "state": IMAGES_STATE} + cases = [ + ["not an object"], + {"page": "images", "name": "Cats"}, + {**good, "extra": 1}, + {**good, "page": "history"}, + {**good, "page": 1}, + {**good, "name": ""}, + {**good, "name": " "}, + {**good, "name": "x" * 101}, + {**good, "name": 7}, + {**good, "name": "bad\nname"}, + {**good, "state": "q=cat"}, + {**good, "state": None}, + {**good, "state": {**IMAGES_STATE, "unknown": ""}}, + {**good, "state": {k: v for k, v in IMAGES_STATE.items() if k != "q"}}, + {**good, "state": {**IMAGES_STATE, "q": 5}}, + {**good, "state": {**IMAGES_STATE, "q": "x" * 501}}, + {**good, "state": {**IMAGES_STATE, "category": "x" * 101}}, + {**good, "state": {**IMAGES_STATE, "sort": "random"}}, + {**good, "state": {**IMAGES_STATE, "media": "audio"}}, + {**good, "state": {**IMAGES_STATE, "source": "-1"}}, + {**good, "state": {**IMAGES_STATE, "source": 0}}, + {**good, "state": {**IMAGES_STATE, "folder": None}}, + {**good, "state": {**IMAGES_STATE, "folder": "x" * 1025}}, + {**good, "state": {**IMAGES_STATE, "tags": "cat"}}, + {**good, "state": {**IMAGES_STATE, "tags": [1]}}, + {**good, "state": {**IMAGES_STATE, "tags": [" "]}}, + {**good, "state": {**IMAGES_STATE, "tags": ["a,b"]}}, + {**good, "state": {**IMAGES_STATE, "tags": ["x" * 51]}}, + {**good, "state": {**IMAGES_STATE, "tags": [f"t{n}" for n in range(21)]}}, + # Library has no source, folder or media, and its own sorts. + {"page": "library", "name": "L", "state": IMAGES_STATE}, + { + "page": "library", + "name": "L", + "state": {**LIBRARY_STATE, "sort": "name-desc"}, + }, + ] + for body in cases: + with self.subTest(body=body): + response, data = await self.send("POST", body=body) + self.assertEqual(response.status, 400) + self.assertFalse(data["success"]) + self.assertIsInstance(data["error"], str) + response = await self.client.post(URL, data="{not json") + self.assertEqual(response.status, 400) + self.assertFalse((await response.json())["success"]) + _, listed = await self.send("GET") + self.assertEqual(listed["views"], []) + + async def test_encoded_state_is_capped(self): + # Each value is within its own limit; together they pass 4 KB. + state = { + **IMAGES_STATE, + "q": "é" * 500, + "folder": "é" * 1024, + "tags": ["é" * 50 for _ in range(20)], + } + encoded = json.dumps(state, ensure_ascii=False, separators=(",", ":")) + self.assertGreater(len(encoded.encode("utf-8")), MAX_STATE_BYTES) + response, data = await self.send( + "POST", body={"page": "images", "name": "Big", "state": state} + ) + self.assertEqual(response.status, 400) + self.assertIn("4 KB", data["error"]) + + async def test_every_documented_value_is_accepted(self): + for sort in ("newest", "oldest", "name", "name-desc", "rating"): + for media in ("all", "image", "video"): + await self.create(state={"sort": sort, "media": media, "source": "3"}) + for sort in ("newest", "oldest", "name", "rating"): + await self.create(page="library", state={"sort": sort, "tags": ["a b"]}) + _, listed = await self.send("GET") + self.assertEqual(len(listed["views"]), 19) + + async def test_rename_and_delete_validate_id_and_body(self): + view = await self.create() + for path in ("/0", "/-1", "/abc", "/1.5", f"/{2**63}"): + for method, body in (("PATCH", {"name": "X"}), ("DELETE", None)): + with self.subTest(path=path, method=method): + response, data = await self.send(method, path, body) + self.assertEqual(response.status, 400) + self.assertFalse(data["success"]) + for body in ({}, {"name": ""}, {"name": "X", "page": "library"}, ["X"]): + with self.subTest(body=body): + response, _ = await self.send("PATCH", f"/{view['id']}", body) + self.assertEqual(response.status, 400) + _, listed = await self.send("GET") + self.assertEqual(listed["views"], [view]) + + async def test_database_failure_is_a_500_envelope(self): + def broken(): + raise sqlite3.OperationalError("disk I/O error") + + self.api.db.list_saved_views = broken + with self.assertLogs("prompt_manager.api", level="ERROR"): + response, data = await self.send("GET") + self.assertEqual(response.status, 500) + self.assertEqual( + data, {"success": False, "error": "Saved view operation failed"} + ) + + +class TestSavedViewsSchema(unittest.TestCase): + def setUp(self): + self.tmpdir = tempfile.mkdtemp() + self.path = os.path.join(self.tmpdir, "prompts.db") + self.db = PromptDatabase(self.path) + self.addCleanup(self._cleanup) + + def _cleanup(self): + PromptModel.close_all_instances() + assert_closed(self, self.tmpdir) + for name in os.listdir(self.tmpdir): + os.unlink(os.path.join(self.tmpdir, name)) + os.rmdir(self.tmpdir) + + def _source(self, name="source.db", views=()): + """A closed database file holding a prompt and the given views.""" + path = os.path.join(self.tmpdir, name) + src = PromptDatabase(path) + try: + src.save_prompt(text="kept", prompt_hash=generate_prompt_hash("kept")) + for page, view_name, state in views: + src.create_saved_view(page, view_name, state) + finally: + src.close_all() + return path + + def test_current_database_has_a_valid_table(self): + conn = self.db.model.get_connection() + self.assertIsNone(saved_views_schema_error(conn, allow_missing=False)) + ok, reason = PromptModel.verify_database_file( + self.path, require_library_state=True + ) + self.assertTrue(ok, reason) + + def test_migration_is_idempotent(self): + conn = self.db.model.get_connection() + self.db.create_saved_view("images", "Cats", IMAGES_STATE) + migrate_saved_views(conn) + self.assertEqual(len(self.db.list_saved_views()), 1) + + def test_backup_restore_round_trip_keeps_saved_views(self): + self.db.create_saved_view("library", "Live", LIBRARY_STATE) + src_path = self._source( + views=[ + ("images", "Cats", IMAGES_STATE), + ("library", "Best", LIBRARY_STATE), + ] + ) + # The views may still sit in the WAL; the backup must carry them. + backup = os.path.join(self.tmpdir, "backup.db") + reopened = PromptDatabase(src_path) + try: + self.assertTrue(reopened.model.backup_database(backup)) + finally: + reopened.close_all() + + previous = self.db.model.restore_from_file(backup) + + self.assertEqual( + [ + (view["page"], view["name"], view["state"]) + for view in self.db.list_saved_views() + ], + [("images", "Cats", IMAGES_STATE), ("library", "Best", LIBRARY_STATE)], + ) + created = self.db.create_saved_view("images", "After", IMAGES_STATE) + self.assertEqual(self.db.list_saved_views()[-1], created) + with closing(sqlite3.connect(previous)) as conn: + self.assertEqual( + conn.execute(f"SELECT name FROM {TABLE}").fetchall(), [("Live",)] + ) + + def test_restore_old_backup_without_saved_views(self): + self.db.create_saved_view("images", "Live", IMAGES_STATE) + src = self._source() + with closing(sqlite3.connect(src)) as conn: + conn.execute(f"DROP TABLE {TABLE}") + conn.commit() + self.assertTrue(PromptModel.verify_database_file(src)[0]) + ok, _ = PromptModel.verify_database_file(src, require_library_state=True) + self.assertFalse(ok) + + self.db.model.restore_from_file(src) + + self.assertEqual(self.db.list_saved_views(), []) + self.db.create_saved_view("library", "New", LIBRARY_STATE) + self.assertEqual(len(self.db.list_saved_views()), 1) + with closing(sqlite3.connect(src)) as conn: # the source is untouched + self.assertIsNone( + conn.execute( + "SELECT 1 FROM sqlite_master WHERE name=?", (TABLE,) + ).fetchone() + ) + + def _tampered(self, *statements, views=(("images", "Cats", IMAGES_STATE),)): + src = self._source( + name=f"tampered{len(os.listdir(self.tmpdir))}.db", views=views + ) + with closing(sqlite3.connect(src)) as conn: + for statement in statements: + conn.execute(statement) + conn.commit() + return src + + def test_restore_rejects_tampered_saved_views(self): + bad_state = json.dumps({**IMAGES_STATE, "unknown": 1}) + cases = { + "definition": ( + f"DROP TABLE {TABLE}", + f"CREATE TABLE {TABLE} (id INTEGER PRIMARY KEY, page TEXT, " + "name TEXT, state TEXT, position INTEGER, created_at TEXT, " + "updated_at TEXT)", + ), + "index": (f"CREATE INDEX saved_views_extra ON {TABLE}(name)",), + "page": ( + "PRAGMA ignore_check_constraints = ON", + f"UPDATE {TABLE} SET page = 'history'", + ), + "name": (f"UPDATE {TABLE} SET name = ''",), + "long name": (f"UPDATE {TABLE} SET name = '{'x' * 101}'",), + "untrimmed": (f"UPDATE {TABLE} SET name = ' Cats'",), + "blob name": (f"UPDATE {TABLE} SET name = CAST('Cats' AS BLOB)",), + "state json": (f"UPDATE {TABLE} SET state = '{{'",), + "state keys": (f"UPDATE {TABLE} SET state = '{bad_state}'",), + "state size": ( + f"UPDATE {TABLE} SET state = '{'x' * (MAX_STATE_BYTES + 1)}'", + ), + "position": (f"UPDATE {TABLE} SET position = 'first'",), + "timestamp": (f"UPDATE {TABLE} SET created_at = X'00'",), + } + for label, statements in cases.items(): + with self.subTest(label=label): + src = self._tampered(*statements) + ok, reason = PromptModel.verify_database_file(src) + self.assertFalse(ok) + self.assertIn("Saved views", reason) + with self.assertRaises(ValueError): + self.db.model.restore_from_file(src) + + def test_migration_refuses_an_unsupported_table(self): + path = os.path.join(self.tmpdir, "foreign.db") + with closing(sqlite3.connect(path)) as conn: + conn.execute(f"CREATE TABLE {TABLE} (id INTEGER PRIMARY KEY)") + with self.assertRaises(sqlite3.DatabaseError): + migrate_saved_views(conn) + # The savepoint rolled back, so the foreign table is untouched. + self.assertEqual( + conn.execute( + f"SELECT COUNT(*) FROM pragma_table_info('{TABLE}')" + ).fetchone()[0], + 1, + ) + self.assertTrue(Path(path).exists()) + + +if __name__ == "__main__": + unittest.main()