Validate export API JSON body and use tempfile paths in engine tests

Author: clean6378-max-it

api/export_api.py

@@ -93,7 +93,9 @@ def export_chats() -> tuple[Response, int] | Response:
     exclusion rules file.
     """
     try:
-        body = request.get_json(silent=True) or {}
+        body = request.get_json(silent=True)
+        if not isinstance(body, dict):
+            return json_response({"error": "request body must be a JSON object"}, 400)
         since: Literal["all", "last"] = (
             "last" if body.get("since") == "last" else "all"
         )

tests/test_api_export.py

@@ -91,6 +91,15 @@ def test_post_returns_zip_with_markdown_entry(self, client, export_state_dir):
 
 
 class TestExportErrorResponses:
+    def test_non_dict_json_body_returns_400(self, client, export_state_dir):
+        response = client.post(
+            "/api/export",
+            json=["not", "an", "object"],
+            content_type="application/json",
+        )
+        assert response.status_code == 400
+        assert response.get_json().get("error") == "request body must be a JSON object"
+
     def test_missing_global_storage_returns_404(self, empty_workspace_client):
         response = _post_export(empty_workspace_client)
         assert response.status_code == 404

tests/test_export_engine.py

@@ -4,6 +4,7 @@
 
 import os
 import sys
+import tempfile
 import unittest
 from unittest.mock import MagicMock, patch
 
@@ -18,7 +19,17 @@
 )
 
 
-class TestCollectExportEntriesNocache(unittest.TestCase):
+class _TempExportPathsMixin:
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self._tmp.cleanup)
+        self.tmp_ws = os.path.join(self._tmp.name, "ws")
+        self.tmp_out = os.path.join(self._tmp.name, "out")
+        os.makedirs(self.tmp_ws, exist_ok=True)
+        os.makedirs(self.tmp_out, exist_ok=True)
+
+
+class TestCollectExportEntriesNocache(_TempExportPathsMixin, unittest.TestCase):
     def test_nocache_env_passed_to_prepare_workspace_orchestration(self):
         with patch.dict(os.environ, {"CURSOR_CHAT_BROWSER_NOCACHE": "1"}):
             with patch(
@@ -30,27 +41,30 @@ def test_nocache_env_passed_to_prepare_workspace_orchestration(self):
                     return_value=None,
                 ):
                     collect_export_entries(
-                        workspace_path="/tmp/ws",
+                        workspace_path=self.tmp_ws,
                         exclusion_rules=[],
                         since="all",
                         last_export_ms=0,
-                        out_dir="/tmp/out",
+                        out_dir=self.tmp_out,
                         include_composer=False,
                         include_cli=False,
                     )
         mock_prepare.assert_called_once()
         self.assertTrue(mock_prepare.call_args.kwargs["nocache"])
 
 
-class TestCollectExportEntriesCorruptComposer(unittest.TestCase):
+class TestCollectExportEntriesCorruptComposer(
+    _TempExportPathsMixin,
+    unittest.TestCase,
+):
     def test_non_dict_composer_row_is_skipped(self):
         ctx = MagicMock()
         ctx.project_name_to_workspace_id = {}
         ctx.workspace_path_to_id = {}
         ctx.composer_id_to_workspace_id = {}
         ctx.invalid_workspace_ids = set()
         orch = WorkspaceOrchestration(
-            workspace_path="/tmp/ws",
+            workspace_path=self.tmp_ws,
             workspace_entries=[],
             fingerprint={},
             ctx=ctx,
@@ -80,11 +94,11 @@ def __getitem__(self, key: str) -> str:
                 return_value=db_data,
             ):
                 exported = collect_export_entries(
-                    workspace_path="/tmp/ws",
+                    workspace_path=self.tmp_ws,
                     exclusion_rules=[],
                     since="all",
                     last_export_ms=0,
-                    out_dir="/tmp/out",
+                    out_dir=self.tmp_out,
                     include_cli=False,
                 )
         self.assertEqual(exported, [])