feat: upload - deel 03.02 - Skipp all toegevoegd

webui/backend/app/api/routes_files.py

@@ -37,10 +37,11 @@ async def delete(
 @router.post("/upload", response_model=UploadResponse)
 async def upload(
     target_path: str = Form(...),
+    overwrite: bool = Form(False),
     file: UploadFile = File(...),
     service: FileOpsService = Depends(get_file_ops_service),
 ) -> UploadResponse:
-    return service.upload(target_path=target_path, upload_file=file)
+    return service.upload(target_path=target_path, upload_file=file, overwrite=overwrite)
 
 
 @router.get("/view", response_model=ViewResponse)

webui/backend/app/fs/filesystem_adapter.py

@@ -140,8 +140,9 @@ class FilesystemAdapter:
             "modified": self.modified_iso(path),
         }
 
-    def write_uploaded_file(self, path: Path, file_stream, chunk_size: int = 1024 * 1024) -> dict:
-        with path.open("xb") as handle:
+    def write_uploaded_file(self, path: Path, file_stream, chunk_size: int = 1024 * 1024, overwrite: bool = False) -> dict:
+        mode = "wb" if overwrite else "xb"
+        with path.open(mode) as handle:
             while True:
                 chunk = file_stream.read(chunk_size)
                 if not chunk:

webui/backend/app/services/file_ops_service.py

@@ -204,7 +204,7 @@ class FileOpsService:
             self._record_history_error(operation="delete", path=path, error=error)
             raise error
 
-    def upload(self, target_path: str, upload_file) -> UploadResponse:
+    def upload(self, target_path: str, upload_file, overwrite: bool = False) -> UploadResponse:
         destination_relative = None
         history_path = target_path
         try:
@@ -216,14 +216,26 @@ class FileOpsService:
             resolved_destination = self._path_guard.resolve_path(destination_relative)
 
             if resolved_destination.absolute.exists():
-                raise AppError(
-                    code="already_exists",
-                    message="Target path already exists",
-                    status_code=409,
-                    details={"path": resolved_destination.relative},
-                )
+                if not overwrite:
+                    raise AppError(
+                        code="already_exists",
+                        message="Target path already exists",
+                        status_code=409,
+                        details={"path": resolved_destination.relative},
+                    )
+                if resolved_destination.absolute.is_dir():
+                    raise AppError(
+                        code="type_conflict",
+                        message="Cannot overwrite an existing directory",
+                        status_code=409,
+                        details={"path": resolved_destination.relative},
+                    )
 
-            saved = self._filesystem.write_uploaded_file(resolved_destination.absolute, upload_file.file)
+            saved = self._filesystem.write_uploaded_file(
+                resolved_destination.absolute,
+                upload_file.file,
+                overwrite=overwrite,
+            )
             self._record_history(
                 operation="upload",
                 status="completed",

webui/backend/tests/golden/test_api_upload_golden.py

@@ -49,13 +49,13 @@ class UploadApiGoldenTest(unittest.TestCase):
         app.dependency_overrides.clear()
         self.temp_dir.cleanup()
 
-    def _upload(self, *, target_path: str, filename: str, content: bytes) -> httpx.Response:
+    def _upload(self, *, target_path: str, filename: str, content: bytes, overwrite: bool = False) -> httpx.Response:
         async def _run() -> httpx.Response:
             transport = httpx.ASGITransport(app=app)
             async with httpx.AsyncClient(transport=transport, base_url="http://testserver") as client:
                 return await client.post(
                     "/api/files/upload",
-                    data={"target_path": target_path},
+                    data={"target_path": target_path, "overwrite": "true" if overwrite else "false"},
                     files={"file": (filename, content, "application/octet-stream")},
                 )
 
@@ -184,3 +184,21 @@ class UploadApiGoldenTest(unittest.TestCase):
         self.assertEqual(history[0]["operation"], "upload")
         self.assertEqual(history[0]["status"], "failed")
         self.assertEqual(history[0]["error_code"], "already_exists")
+
+    def test_upload_overwrite_existing_file_success(self) -> None:
+        existing = self.uploads_dir / "hello.txt"
+        existing.write_text("existing", encoding="utf-8")
+
+        response = self._upload(
+            target_path="storage1/uploads",
+            filename="hello.txt",
+            content=b"replacement",
+            overwrite=True,
+        )
+
+        self.assertEqual(response.status_code, 200)
+        self.assertEqual((self.uploads_dir / "hello.txt").read_bytes(), b"replacement")
+
+        history = self._get_history()
+        self.assertEqual(history[0]["operation"], "upload")
+        self.assertEqual(history[0]["status"], "completed")

webui/backend/tests/golden/test_ui_smoke_golden.py

@@ -203,6 +203,19 @@ class UiSmokeGoldenTest(unittest.TestCase):
         self.assertIn('async function handleUploadSelection(event)', app_js)
         self.assertIn('uploadElements().input.onchange = handleUploadSelection;', app_js)
         self.assertIn('"/api/files/upload"', app_js)
+        self.assertIn('function ensureUploadConflictModal()', app_js)
+        self.assertIn('function promptUploadConflict(', app_js)
+        self.assertIn('formData.append("overwrite", overwrite ? "true" : "false")', app_js)
+        self.assertIn('createButton("Overwrite"', app_js)
+        self.assertIn('createButton("Overwrite all"', app_js)
+        self.assertIn('createButton("Skip"', app_js)
+        self.assertIn('createButton("Skip all"', app_js)
+        self.assertIn('createButton("Cancel"', app_js)
+        self.assertIn('if (err.code !== "already_exists") {', app_js)
+        self.assertIn('if (choice === "overwrite_all") {', app_js)
+        self.assertIn('if (uploadState.skipAll) {', app_js)
+        self.assertIn('if (choice === "skip_all") {', app_js)
+        self.assertIn('uploadState.skipAll = true;', app_js)
         self.assertIn('Upload to: ${uploadState.targetPath}', app_js)
         self.assertIn('Uploading ${total} file', app_js)
         self.assertIn('`/api/files/thumbnail?', app_js)