|
|
@@ -20,7 +20,7 @@ from unittest.mock import Mock, patch
|
|
|
|
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
|
|
-from filebrowser.api import get_all_filesystems, mkdir, move, rename, upload_file
|
|
|
+from filebrowser.api import copy, get_all_filesystems, mkdir, move, rename, upload_file
|
|
|
from filebrowser.conf import (
|
|
|
MAX_FILE_SIZE_UPLOAD_LIMIT,
|
|
|
RESTRICT_FILE_EXTENSIONS,
|
|
|
@@ -725,3 +725,223 @@ class TestGetFilesystemsAPI:
|
|
|
{'file_system': 's3a', 'user_home_directory': 's3a://test-bucket/test-user-home-dir/', 'config': {}},
|
|
|
{'file_system': 'ofs', 'user_home_directory': 'ofs://', 'config': {}},
|
|
|
]
|
|
|
+
|
|
|
+
|
|
|
+class TestCopyAPI:
|
|
|
+ def test_copy_normal_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',
|
|
|
+ ]
|
|
|
+ ),
|
|
|
+ copy=Mock(),
|
|
|
+ user=Mock(),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ response = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 200
|
|
|
+ request.fs.copy.assert_called_once_with(
|
|
|
+ 's3a://test-bucket/test-user/src_dir/source.txt', 's3a://test-bucket/test-user/dst_dir', recursive=True, owner=request.user
|
|
|
+ )
|
|
|
+
|
|
|
+ def test_copy_ofs_success(self):
|
|
|
+ request = Mock(
|
|
|
+ method='POST',
|
|
|
+ POST={
|
|
|
+ 'source_path': 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt',
|
|
|
+ 'destination_path': 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ },
|
|
|
+ fs=Mock(
|
|
|
+ exists=Mock(side_effect=[True, False]),
|
|
|
+ isdir=Mock(return_value=True),
|
|
|
+ parent_path=Mock(return_value='ofs://test_vol/test-bucket/test-user/src_dir'),
|
|
|
+ join=Mock(return_value='ofs://test_vol/test-bucket/test-user/dst_dir/source.txt'),
|
|
|
+ normpath=Mock(
|
|
|
+ side_effect=[
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt',
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ ]
|
|
|
+ ),
|
|
|
+ copy=Mock(return_value=''),
|
|
|
+ user=Mock(),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ response = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 200
|
|
|
+ request.fs.copy.assert_called_once_with(
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt',
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ recursive=True,
|
|
|
+ owner=request.user,
|
|
|
+ )
|
|
|
+
|
|
|
+ def test_copy_ofs_skip_files_error(self):
|
|
|
+ request = Mock(
|
|
|
+ method='POST',
|
|
|
+ POST={
|
|
|
+ 'source_path': 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt',
|
|
|
+ 'destination_path': 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ },
|
|
|
+ fs=Mock(
|
|
|
+ exists=Mock(side_effect=[True, False]),
|
|
|
+ isdir=Mock(return_value=True),
|
|
|
+ parent_path=Mock(return_value='ofs://test_vol/test-bucket/test-user/src_dir'),
|
|
|
+ join=Mock(return_value='ofs://test_vol/test-bucket/test-user/dst_dir/source.txt'),
|
|
|
+ normpath=Mock(
|
|
|
+ side_effect=[
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt',
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ ]
|
|
|
+ ),
|
|
|
+ copy=Mock(return_value=('ofs://test_vol/test-bucket/test-user/src_dir/source.txt')),
|
|
|
+ user=Mock(),
|
|
|
+ ),
|
|
|
+ )
|
|
|
+ response = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 500
|
|
|
+ assert json.loads(response.content) == {'skipped_files': 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt'}
|
|
|
+ request.fs.copy.assert_called_once_with(
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/src_dir/source.txt',
|
|
|
+ 'ofs://test_vol/test-bucket/test-user/dst_dir',
|
|
|
+ recursive=True,
|
|
|
+ owner=request.user,
|
|
|
+ )
|
|
|
+
|
|
|
+ def test_copy_no_source_path(self):
|
|
|
+ request = Mock(
|
|
|
+ method='POST',
|
|
|
+ POST={'destination_path': 's3a://test-bucket/test-user/dst_dir'},
|
|
|
+ fs=Mock(),
|
|
|
+ )
|
|
|
+ response = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 400
|
|
|
+ assert response.content.decode('utf-8') == 'Missing required parameters: source_path and destination_path are required.'
|
|
|
+
|
|
|
+ def test_copy_no_destination_path(self):
|
|
|
+ request = Mock(
|
|
|
+ method='POST',
|
|
|
+ POST={'source_path': 's3a://test-bucket/test-user/src_dir/source.txt'},
|
|
|
+ fs=Mock(),
|
|
|
+ )
|
|
|
+ response = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 400
|
|
|
+ assert response.content.decode('utf-8') == 'Missing required parameters: source_path and destination_path are required.'
|
|
|
+
|
|
|
+ def test_copy_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 = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 400
|
|
|
+ assert response.content.decode('utf-8') == 'Source and destination paths must be different.'
|
|
|
+
|
|
|
+ def test_copy_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 = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 404
|
|
|
+ assert response.content.decode('utf-8') == 'Source file or folder does not exist.'
|
|
|
+
|
|
|
+ def test_copy_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 = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 400
|
|
|
+ assert response.content.decode('utf-8') == 'Destination path must be a directory.'
|
|
|
+
|
|
|
+ def test_copy_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 = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 400
|
|
|
+ assert response.content.decode('utf-8') == 'Destination cannot be the parent directory of source.'
|
|
|
+
|
|
|
+ def test_copy_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 = copy(request)
|
|
|
+
|
|
|
+ assert response.status_code == 409
|
|
|
+ assert response.content.decode('utf-8') == 'File or folder already exists at destination path.'
|