ComfyAssets-ComfyUI_PromptManager/tests/test_diagnostics_config.py

"""
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.assertEqual(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()