Vito Sansevero
5ce3349d64
- I1: once the stage's copy replaced the library the restore succeeded: pruning, reopening self.db, dropping old previews and closing the old object only log, so the job finishes and the legacy /restore answers 200 instead of "the library was not changed". The generation is always bumped. - M1: a backup download holds its guarded slot only while the snapshot is written; the stream runs outside it, so a stalled download blocks nothing. - M2: a restore preview leases only its read of the live library, not the upload or the staging; a restore whose drain still times out keeps the reviewed stage under its token. - M3: the reviewed-preview store changes only on the event loop; stage folders are removed in the executor. - M4: a document whose body inflates completely while only its recorded hash or only its length disagrees is re-keyed (or merged into the document holding that text) instead of dropped. - M5: a backup download is refused only when the snapshot is not SQLite or fails integrity_check; a semantic check failure still downloads, logged, sent as X-PromptManager-Verify and recorded as not verified. - M6: a refused, failed or cancelled preview removes its stage, also when the cancel lands while a worker still stages; a timer removes expired stages without another request. - M7, N1: repair reads rows one at a time and counts distinct images. - N3: the live fingerprint never commits a transaction it did not open.
213 lines
8.1 KiB
Python
213 lines
8.1 KiB
Python
"""
|
|
Tests for issue #132: Diagnostics, backup, and restore must respect config.json.
|
|
|
|
Verifies that:
|
|
- GalleryDiagnostics resolves db_path from config (not hardcoded "prompts.db")
|
|
- The admin API diagnostics/backup/restore endpoints use the configured db_path
|
|
- Output directory checks consult GalleryConfig before falling back to heuristics
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
from contextlib import closing
|
|
import sys
|
|
import tempfile
|
|
import types
|
|
import unittest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
EXTENSION_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GalleryDiagnostics path resolution
|
|
# ---------------------------------------------------------------------------
|
|
class TestGalleryDiagnosticsDbPath(unittest.TestCase):
|
|
"""GalleryDiagnostics should resolve db_path from config, not hardcode it."""
|
|
|
|
def test_default_resolves_from_config(self):
|
|
"""GalleryDiagnostics() with no args should read config like PromptDatabase."""
|
|
from utils.diagnostics import GalleryDiagnostics
|
|
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
custom_path = os.path.join(tmpdir, "custom.db")
|
|
fake_config_mod = types.ModuleType("py.config")
|
|
fake_pm_config = type(
|
|
"PromptManagerConfig", (), {"DEFAULT_DB_PATH": custom_path}
|
|
)
|
|
fake_config_mod.PromptManagerConfig = fake_pm_config
|
|
|
|
with patch.dict(sys.modules, {"py.config": fake_config_mod}):
|
|
diag = GalleryDiagnostics()
|
|
|
|
self.assertEqual(diag.db_path, custom_path)
|
|
|
|
def test_default_without_config_falls_back_to_extension_root(self):
|
|
"""Without config, should default to prompts.db in extension root."""
|
|
from utils.diagnostics import GalleryDiagnostics
|
|
|
|
diag = GalleryDiagnostics()
|
|
expected = os.path.join(EXTENSION_ROOT, "prompts.db")
|
|
self.assertEqual(diag.db_path, expected)
|
|
|
|
def test_explicit_path_overrides_config(self):
|
|
"""An explicit db_path argument should be used as-is."""
|
|
from utils.diagnostics import GalleryDiagnostics
|
|
|
|
diag = GalleryDiagnostics("/explicit/path.db")
|
|
self.assertEqual(diag.db_path, "/explicit/path.db")
|
|
|
|
def test_config_relative_path_resolved_to_extension_root(self):
|
|
"""A relative path from config should resolve against extension root."""
|
|
from utils.diagnostics import GalleryDiagnostics
|
|
|
|
fake_config_mod = types.ModuleType("py.config")
|
|
fake_pm_config = type(
|
|
"PromptManagerConfig", (), {"DEFAULT_DB_PATH": "data/prompts.db"}
|
|
)
|
|
fake_config_mod.PromptManagerConfig = fake_pm_config
|
|
|
|
with patch.dict(sys.modules, {"py.config": fake_config_mod}):
|
|
diag = GalleryDiagnostics()
|
|
|
|
expected = os.path.join(EXTENSION_ROOT, "data", "prompts.db")
|
|
self.assertEqual(diag.db_path, expected)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin API endpoints use self.db.model.db_path
|
|
# ---------------------------------------------------------------------------
|
|
class TestAdminEndpointsUseConfigPath(unittest.TestCase):
|
|
"""Admin diagnostics, backup, and restore must use the configured db_path."""
|
|
|
|
def _make_api_stub(self, db_path):
|
|
"""Create a minimal object that mimics the admin mixin's self."""
|
|
from py.api import PromptManagerAPI
|
|
|
|
stub = MagicMock()
|
|
stub.db.model.db_path = db_path
|
|
stub.logger = MagicMock()
|
|
stub._public_path = PromptManagerAPI._public_path
|
|
stub._public_error = PromptManagerAPI._public_error
|
|
|
|
async def run_in_executor(func, *args, **kwargs):
|
|
return func(*args, **kwargs)
|
|
|
|
stub._run_in_executor = run_in_executor
|
|
return stub
|
|
|
|
def _run_async(self, coro):
|
|
"""Run a coroutine on a fresh event loop to avoid aiohttp pollution."""
|
|
loop = asyncio.new_event_loop()
|
|
try:
|
|
return loop.run_until_complete(coro)
|
|
finally:
|
|
loop.close()
|
|
|
|
def test_run_diagnostics_uses_model_db_path(self):
|
|
"""run_diagnostics should check self.db.model.db_path, not 'prompts.db'."""
|
|
if "server" not in sys.modules:
|
|
sys.modules["server"] = MagicMock()
|
|
|
|
from py.api.admin import AdminRoutesMixin
|
|
|
|
with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
|
|
db_file = f.name
|
|
|
|
try:
|
|
import sqlite3
|
|
|
|
with closing(sqlite3.connect(db_file)) as conn, conn:
|
|
conn.execute(
|
|
"CREATE TABLE prompts "
|
|
"(id INTEGER PRIMARY KEY, text TEXT, created_at TEXT)"
|
|
)
|
|
conn.execute(
|
|
"INSERT INTO prompts (text, created_at) "
|
|
"VALUES ('test', '2024-01-01')"
|
|
)
|
|
|
|
stub = self._make_api_stub(db_file)
|
|
result = self._run_async(
|
|
AdminRoutesMixin.run_diagnostics(stub, MagicMock())
|
|
)
|
|
|
|
body = json.loads(result.body)
|
|
self.assertTrue(body["success"])
|
|
self.assertEqual(body["diagnostics"]["database"]["status"], "ok")
|
|
self.assertEqual(body["diagnostics"]["database"]["prompt_count"], 1)
|
|
finally:
|
|
os.unlink(db_file)
|
|
|
|
def test_run_diagnostics_reports_missing_custom_path(self):
|
|
"""If the configured db_path doesn't exist, diagnostics should report error."""
|
|
if "server" not in sys.modules:
|
|
sys.modules["server"] = MagicMock()
|
|
|
|
from py.api.admin import AdminRoutesMixin
|
|
|
|
stub = self._make_api_stub("/nonexistent/custom/prompts.db")
|
|
|
|
result = self._run_async(AdminRoutesMixin.run_diagnostics(stub, MagicMock()))
|
|
body = json.loads(result.body)
|
|
self.assertTrue(body["success"])
|
|
self.assertEqual(body["diagnostics"]["database"]["status"], "error")
|
|
# The message names the file but never the absolute server path.
|
|
message = body["diagnostics"]["database"]["message"]
|
|
self.assertIn("prompts.db", message)
|
|
self.assertNotIn("/nonexistent/custom", message)
|
|
|
|
def test_backup_uses_model_db_path(self):
|
|
"""The backup snapshot is taken of self.db.model.db_path."""
|
|
if "server" not in sys.modules:
|
|
sys.modules["server"] = MagicMock()
|
|
|
|
from py.api.backup_restore import write_backup_snapshot
|
|
|
|
from database.operations import PromptDatabase
|
|
|
|
folder = tempfile.mkdtemp()
|
|
db_file = os.path.join(folder, "custom.db")
|
|
# A WAL-safe backup copies pages through sqlite itself, so the
|
|
# configured path has to hold a real database, not arbitrary bytes.
|
|
real_db = PromptDatabase(db_file)
|
|
try:
|
|
stub = self._make_api_stub(db_file)
|
|
stub.db.model = real_db.model
|
|
snapshot, warning = write_backup_snapshot(stub.db.model)
|
|
try:
|
|
self.assertIsNone(warning)
|
|
self.assertEqual(os.path.dirname(snapshot), folder)
|
|
with open(snapshot, "rb") as handle:
|
|
self.assertTrue(handle.read(16) == b"SQLite format 3\x00")
|
|
finally:
|
|
os.unlink(snapshot)
|
|
finally:
|
|
real_db.close_all()
|
|
for name in os.listdir(folder):
|
|
os.unlink(os.path.join(folder, name))
|
|
os.rmdir(folder)
|
|
|
|
def test_backup_reports_missing_custom_path(self):
|
|
"""The backup answers 404 when the configured path doesn't exist."""
|
|
if "server" not in sys.modules:
|
|
sys.modules["server"] = MagicMock()
|
|
|
|
from py.api.backup_restore import BackupRestoreRoutesMixin
|
|
|
|
stub = self._make_api_stub("/nonexistent/custom/prompts.db")
|
|
|
|
result = self._run_async(
|
|
BackupRestoreRoutesMixin._stream_backup(stub, MagicMock())
|
|
)
|
|
body = json.loads(result.body)
|
|
self.assertFalse(body["success"])
|
|
self.assertEqual(result.status, 404)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|