test(system): a restore refuses a running health check

The health route test relied on a legacy restore interleaving with a running
check; the restore is now exclusive and refuses it, naming the check. The
diagnostics backup tests follow the backup handler into
py/api/backup_restore.py.

tests/test_diagnostics_config.py

@@ -161,46 +161,47 @@ class TestAdminEndpointsUseConfigPath(unittest.TestCase):
         self.assertNotIn("/nonexistent/custom", message)
 
     def test_backup_uses_model_db_path(self):
-        """backup_database should read from self.db.model.db_path."""
+        """The backup snapshot is taken of self.db.model.db_path."""
         if "server" not in sys.modules:
             sys.modules["server"] = MagicMock()
 
-        from py.api.admin import AdminRoutesMixin
+        from py.api.backup_restore import write_backup_snapshot
 
         from database.operations import PromptDatabase
 
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
-            db_file = f.name
-
+        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
-            result = self._run_async(
-                AdminRoutesMixin.backup_database(stub, MagicMock())
-            )
-
-            self.assertEqual(result.content_type, "application/octet-stream")
-            self.assertGreater(len(result.body), 0)
-            self.assertTrue(result.body.startswith(b"SQLite format 3\x00"))
+            snapshot = write_backup_snapshot(stub.db.model)
+            try:
+                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 suffix in ("", "-wal", "-shm"):
-                if os.path.exists(db_file + suffix):
-                    os.unlink(db_file + suffix)
+            for name in os.listdir(folder):
+                os.unlink(os.path.join(folder, name))
+            os.rmdir(folder)
 
     def test_backup_reports_missing_custom_path(self):
-        """backup_database should return 404 when configured path doesn't exist."""
+        """The backup answers 404 when the configured path doesn't exist."""
         if "server" not in sys.modules:
             sys.modules["server"] = MagicMock()
 
-        from py.api.admin import AdminRoutesMixin
+        from py.api.backup_restore import BackupRestoreRoutesMixin
 
         stub = self._make_api_stub("/nonexistent/custom/prompts.db")
 
-        result = self._run_async(AdminRoutesMixin.backup_database(stub, MagicMock()))
+        result = self._run_async(
+            BackupRestoreRoutesMixin._stream_backup(stub, MagicMock())
+        )
         body = json.loads(result.body)
         self.assertFalse(body["success"])
         self.assertEqual(result.status, 404)

tests/test_system_api.py

@@ -60,8 +60,8 @@ class SystemAPITestCase(AdminAPITestCase):
         self.assertEqual(response.status, status, await response.text())
         return await response.json()
 
-    async def restore_snapshot(self):
-        """Restore the live database from a copy of itself through the legacy route."""
+    def snapshot_form(self):
+        """A multipart upload of a copy of the live database."""
         from contextlib import closing
 
         from aiohttp import FormData
@@ -74,8 +74,12 @@ class SystemAPITestCase(AdminAPITestCase):
             upload = handle.read()
         form = FormData()
         form.add_field("database_file", upload, filename="upload.db")
+        return form
+
+    async def restore_snapshot(self):
+        """Restore the live database from a copy of itself through the legacy route."""
         response = await self.client.request(
-            "POST", "/prompt_manager/restore", data=form
+            "POST", "/prompt_manager/restore", data=self.snapshot_form()
         )
         self.assertEqual(response.status, 200, await response.text())
 
@@ -193,8 +197,10 @@ class TestHealthRoute(SystemAPITestCase):
         overview = await self.json("GET", "/prompt_manager/system/overview")
         self.assertIsNone(overview["health"])
 
-    async def test_check_finishing_after_a_restore_is_not_kept(self):
-        # A check that started on the old database describes it, not the new one.
+    async def test_restore_refused_while_a_check_runs(self):
+        # #249 S8 carry-over: a restore and the health check never interleave
+        # (the restore is exclusive), so no check outlives the database it
+        # read; a restore after the check forgets its report.
         self.save("Any")
         release = threading.Event()
         entered = threading.Event()
@@ -209,18 +215,24 @@ class TestHealthRoute(SystemAPITestCase):
             body = await self.json("POST", "/prompt_manager/system/health", 202)
             loop = asyncio.get_running_loop()
             await loop.run_in_executor(None, entered.wait, WAIT)
-            await self.restore_snapshot()
+            refused = await self.client.post(
+                "/prompt_manager/restore", data=self.snapshot_form()
+            )
+            self.assertEqual(refused.status, 409)
+            self.assertEqual(
+                await refused.json(),
+                {"success": False, "error": "Health check already running"},
+            )
             release.set()
             job = self.api._jobs.get(body["job"]["id"])
             await asyncio.wait_for(asyncio.shield(job.task), WAIT)
         self.assertEqual(job.state, "finished")
-        body = await self.json("GET", "/prompt_manager/system/health")
-        self.assertEqual(body, {"success": True, "report": None})
-        # A check started on the restored database is kept again.
-        await self.run_health()
         self.assertIsNotNone(
             (await self.json("GET", "/prompt_manager/system/health"))["report"]
         )
+        await self.restore_snapshot()
+        body = await self.json("GET", "/prompt_manager/system/health")
+        self.assertEqual(body, {"success": True, "report": None})
 
     async def test_health_sources_resolve_off_the_event_loop(self):
         loop_thread = threading.get_ident()