Browse Source

[api] Refactor /move API with additional validation checks (#3991)

Harsh Gupta 9 months ago
parent
commit
96978e742f

+ 38 - 3
apps/filebrowser/src/filebrowser/api.py

@@ -554,7 +554,7 @@ def mkdir(request):
   path = request.POST.get('path')
   path = request.POST.get('path')
   name = request.POST.get('name')
   name = request.POST.get('name')
 
 
-  # Check if source and destination paths are provided
+  # Check if path and name are provided
   if not path or not name:
   if not path or not name:
     return HttpResponse("Missing required parameters: path and name are required.", status=400)
     return HttpResponse("Missing required parameters: path and name are required.", status=400)
 
 
@@ -647,13 +647,48 @@ def rename(request):
   return HttpResponse(status=200)
   return HttpResponse(status=200)
 
 
 
 
+def _is_destination_parent_of_source(request, source_path, destination_path):
+  """Check if the destination path is the parent directory of the source path."""
+  return request.fs.parent_path(source_path) == request.fs.normpath(destination_path)
+
+
 @api_error_handler
 @api_error_handler
 def move(request):
 def move(request):
+  """
+  Move a file or folder from source path to destination path.
+
+  Args:
+    request: The request object containing source and destination paths
+
+  Returns:
+    Success or error response with appropriate status codes
+  """
   source_path = request.POST.get('source_path', '')
   source_path = request.POST.get('source_path', '')
   destination_path = request.POST.get('destination_path', '')
   destination_path = request.POST.get('destination_path', '')
 
 
-  if source_path == destination_path:
-    return HttpResponse('Source and destination path cannot be same.', status=400)
+  # Check if source and destination paths are provided
+  if not source_path or not destination_path:
+    return HttpResponse("Missing required parameters: source_path and destination_path are required.", status=400)
+
+  # Check if paths are identical
+  if request.fs.normpath(source_path) == request.fs.normpath(destination_path):
+    return HttpResponse('Source and destination paths must be different.', status=400)
+
+  # Verify source path exists
+  if not request.fs.exists(source_path):
+    return HttpResponse('Source file or folder does not exist.', status=404)
+
+  # Check if the destination path is a directory
+  if not request.fs.isdir(destination_path):
+    return HttpResponse('Destination path must be a directory.', status=400)
+
+  # Check if destination path is parent of source path
+  if _is_destination_parent_of_source(request, source_path, destination_path):
+    return HttpResponse('Destination cannot be the parent directory of source.', status=400)
+
+  # Check if file or folder already exists at destination path
+  if request.fs.exists(request.fs.join(destination_path, os.path.basename(source_path))):
+    return HttpResponse('File or folder already exists at destination path.', status=409)
 
 
   request.fs.rename(source_path, destination_path)
   request.fs.rename(source_path, destination_path)
   return HttpResponse(status=200)
   return HttpResponse(status=200)

+ 151 - 1
apps/filebrowser/src/filebrowser/api_test.py

@@ -20,7 +20,7 @@ from unittest.mock import Mock, patch
 
 
 from django.core.files.uploadedfile import SimpleUploadedFile
 from django.core.files.uploadedfile import SimpleUploadedFile
 
 
-from filebrowser.api import get_all_filesystems, mkdir, rename, upload_file
+from filebrowser.api import get_all_filesystems, mkdir, move, rename, upload_file
 from filebrowser.conf import (
 from filebrowser.conf import (
   MAX_FILE_SIZE_UPLOAD_LIMIT,
   MAX_FILE_SIZE_UPLOAD_LIMIT,
   RESTRICT_FILE_EXTENSIONS,
   RESTRICT_FILE_EXTENSIONS,
@@ -515,6 +515,156 @@ class TestRenameAPI:
       reset()
       reset()
 
 
 
 
+class TestMoveAPI:
+  def test_move_success(self):
+    request = Mock(
+      method='POST',
+      POST={'source_path': 's3a://test-bucket/test-user/src_dir/source.txt', 'destination_path': 's3a://test-bucket/test-user/dst_dir'},
+      fs=Mock(
+        exists=Mock(side_effect=[True, False]),
+        isdir=Mock(return_value=True),
+        parent_path=Mock(return_value='s3a://test-bucket/test-user/src_dir'),
+        join=Mock(return_value='s3a://test-bucket/test-user/dst_dir/source.txt'),
+        normpath=Mock(
+          side_effect=[
+            's3a://test-bucket/test-user/src_dir/source.txt',
+            's3a://test-bucket/test-user/dst_dir',
+            's3a://test-bucket/test-user/dst_dir',
+          ]
+        ),
+        rename=Mock(),
+      ),
+    )
+    response = move(request)
+
+    assert response.status_code == 200
+    request.fs.rename.assert_called_once_with('s3a://test-bucket/test-user/src_dir/source.txt', 's3a://test-bucket/test-user/dst_dir')
+
+  def test_move_no_source_path(self):
+    request = Mock(
+      method='POST',
+      POST={'destination_path': 's3a://test-bucket/test-user/dst_dir'},
+      fs=Mock(),
+    )
+    response = move(request)
+
+    assert response.status_code == 400
+    assert response.content.decode('utf-8') == 'Missing required parameters: source_path and destination_path are required.'
+
+  def test_move_no_destination_path(self):
+    request = Mock(
+      method='POST',
+      POST={'source_path': 's3a://test-bucket/test-user/src_dir/source.txt'},
+      fs=Mock(),
+    )
+    response = move(request)
+
+    assert response.status_code == 400
+    assert response.content.decode('utf-8') == 'Missing required parameters: source_path and destination_path are required.'
+
+  def test_move_identical_paths(self):
+    request = Mock(
+      method='POST',
+      POST={
+        'source_path': 's3a://test-bucket/test-user/src_dir/source.txt',
+        'destination_path': 's3a://test-bucket/test-user/src_dir/source.txt',
+      },
+      fs=Mock(
+        normpath=Mock(side_effect=['s3a://test-bucket/test-user/src_dir/source.txt', 's3a://test-bucket/test-user/src_dir/source.txt']),
+      ),
+    )
+    response = move(request)
+
+    assert response.status_code == 400
+    assert response.content.decode('utf-8') == 'Source and destination paths must be different.'
+
+  def test_move_source_path_does_not_exist(self):
+    request = Mock(
+      method='POST',
+      POST={
+        'source_path': 's3a://test-bucket/test-user/src_dir/source.txt',
+        'destination_path': 's3a://test-bucket/test-user/dst_dir',
+      },
+      fs=Mock(
+        exists=Mock(return_value=False),
+        normpath=Mock(side_effect=['s3a://test-bucket/test-user/src_dir/source.txt', 's3a://test-bucket/test-user/dst_dir']),
+      ),
+    )
+    response = move(request)
+
+    assert response.status_code == 404
+    assert response.content.decode('utf-8') == 'Source file or folder does not exist.'
+
+  def test_move_destination_not_a_directory(self):
+    request = Mock(
+      method='POST',
+      POST={
+        'source_path': 's3a://test-bucket/test-user/src_dir/source.txt',
+        'destination_path': 's3a://test-bucket/test-user/dst_dir',
+      },
+      fs=Mock(
+        exists=Mock(return_value=True),
+        isdir=Mock(return_value=False),
+        normpath=Mock(side_effect=['s3a://test-bucket/test-user/src_dir/source.txt', 's3a://test-bucket/test-user/dst_dir']),
+      ),
+    )
+    response = move(request)
+
+    assert response.status_code == 400
+    assert response.content.decode('utf-8') == 'Destination path must be a directory.'
+
+  def test_move_destination_is_parent_of_source(self):
+    request = Mock(
+      method='POST',
+      POST={
+        'source_path': 's3a://test-bucket/test-user/src_dir/source.txt',
+        'destination_path': 's3a://test-bucket/test-user/src_dir',
+      },
+      fs=Mock(
+        exists=Mock(return_value=True),
+        isdir=Mock(return_value=True),
+        parent_path=Mock(return_value='s3a://test-bucket/test-user/src_dir'),
+        normpath=Mock(
+          side_effect=[
+            's3a://test-bucket/test-user/src_dir/source.txt',
+            's3a://test-bucket/test-user/src_dir',
+            's3a://test-bucket/test-user/src_dir',
+          ]
+        ),
+      ),
+    )
+    response = move(request)
+
+    assert response.status_code == 400
+    assert response.content.decode('utf-8') == 'Destination cannot be the parent directory of source.'
+
+  def test_move_file_already_exists_at_destination(self):
+    request = Mock(
+      method='POST',
+      POST={
+        'source_path': 's3a://test-bucket/test-user/src_dir/source.txt',
+        'destination_path': 's3a://test-bucket/test-user/dst_dir',
+      },
+      fs=Mock(
+        exists=Mock(side_effect=[True, True]),
+        isdir=Mock(return_value=True),
+        parent_path=Mock(return_value='s3a://test-bucket/test-user/src_dir'),
+        join=Mock(return_value='s3a://test-bucket/test-user/dst_dir/source.txt'),
+        normpath=Mock(
+          side_effect=[
+            's3a://test-bucket/test-user/src_dir/source.txt',
+            's3a://test-bucket/test-user/dst_dir',
+            's3a://test-bucket/test-user/dst_dir',
+          ]
+        ),
+      ),
+    )
+    response = move(request)
+
+    assert response.status_code == 409
+    assert response.content.decode('utf-8') == 'File or folder already exists at destination path.'
+
+
 class TestGetFilesystemsAPI:
 class TestGetFilesystemsAPI:
   def test_get_all_filesystems_without_hdfs(self):
   def test_get_all_filesystems_without_hdfs(self):
     with patch('filebrowser.api.fsmanager.get_filesystems') as get_filesystems:
     with patch('filebrowser.api.fsmanager.get_filesystems') as get_filesystems: