|
@@ -16,11 +16,12 @@
|
|
|
# limitations under the License.
|
|
# limitations under the License.
|
|
|
|
|
|
|
|
import json
|
|
import json
|
|
|
-from unittest.mock import Mock, patch
|
|
|
|
|
|
|
+from unittest.mock import MagicMock, Mock, patch
|
|
|
|
|
|
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
|
|
|
|
|
|
-from filebrowser.api import copy, get_all_filesystems, mkdir, move, rename, upload_file
|
|
|
|
|
|
|
+from aws.s3.s3fs import S3ListAllBucketsException
|
|
|
|
|
+from filebrowser.api import copy, get_all_filesystems, listdir_paged, 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,
|
|
@@ -945,3 +946,205 @@ class TestCopyAPI:
|
|
|
|
|
|
|
|
assert response.status_code == 409
|
|
assert response.status_code == 409
|
|
|
assert response.content.decode('utf-8') == 'File or folder already exists at destination path.'
|
|
assert response.content.decode('utf-8') == 'File or folder already exists at destination path.'
|
|
|
|
|
+
|
|
|
|
|
+
|
|
|
|
|
+class TestListAPI:
|
|
|
|
|
+ def _create_mock_file_stats(self, name, path, size, user, group):
|
|
|
|
|
+ mock_stats = MagicMock()
|
|
|
|
|
+ mock_stats.path = path
|
|
|
|
|
+ mock_stats.name = name
|
|
|
|
|
+ mock_stats.size = size
|
|
|
|
|
+ mock_stats.atime = 0
|
|
|
|
|
+ mock_stats.mtime = 0
|
|
|
|
|
+ mock_stats.type = 'file'
|
|
|
|
|
+ mock_stats.user = user
|
|
|
|
|
+ mock_stats.group = group
|
|
|
|
|
+ mock_stats.mode = 33188
|
|
|
|
|
+ mock_stats.to_json_dict = Mock(
|
|
|
|
|
+ return_value={
|
|
|
|
|
+ 'path': path,
|
|
|
|
|
+ 'aclBit': False,
|
|
|
|
|
+ 'size': size,
|
|
|
|
|
+ 'atime': 0,
|
|
|
|
|
+ 'mtime': 0,
|
|
|
|
|
+ 'type': 'file',
|
|
|
|
|
+ 'user': user,
|
|
|
|
|
+ 'group': group,
|
|
|
|
|
+ 'mode': 33188,
|
|
|
|
|
+ }
|
|
|
|
|
+ )
|
|
|
|
|
+ return mock_stats
|
|
|
|
|
+
|
|
|
|
|
+ def _create_mock_paginator(self, all_stats):
|
|
|
|
|
+ mock_page = MagicMock()
|
|
|
|
|
+ mock_page.object_list = all_stats
|
|
|
|
|
+ mock_page.has_next.return_value = False
|
|
|
|
|
+ mock_page.has_previous.return_value = False
|
|
|
|
|
+ mock_page.number = 1
|
|
|
|
|
+
|
|
|
|
|
+ mock_paginator = MagicMock()
|
|
|
|
|
+ mock_paginator.page.return_value = mock_page
|
|
|
|
|
+ mock_paginator.per_page = 30
|
|
|
|
|
+ mock_paginator.count = 2
|
|
|
|
|
+ mock_paginator.num_pages = 1
|
|
|
|
|
+
|
|
|
|
|
+ return mock_paginator
|
|
|
|
|
+
|
|
|
|
|
+ def test_listdir_paged_success(self):
|
|
|
|
|
+ with patch('filebrowser.api.is_admin') as is_admin:
|
|
|
|
|
+ is_admin.return_value = False
|
|
|
|
|
+
|
|
|
|
|
+ file1_stats = self._create_mock_file_stats(
|
|
|
|
|
+ name='file1.txt', path='s3a://test-bucket/test-user/test-dir/file1.txt', size=100, user='user1', group='group1'
|
|
|
|
|
+ )
|
|
|
|
|
+ file2_stats = self._create_mock_file_stats(
|
|
|
|
|
+ name='file2.txt', path='s3a://test-bucket/test-user/test-dir/file2.txt', size=200, user='user2', group='group2'
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ all_stats = [file2_stats, file1_stats]
|
|
|
|
|
+
|
|
|
|
|
+ request = Mock(
|
|
|
|
|
+ method='GET',
|
|
|
|
|
+ GET={'pagenum': '1', 'pagesize': '30', 'path': 's3a://test-bucket/test-user/test-dir', 'sortby': 'name', 'descending': 'False'},
|
|
|
|
|
+ user=Mock(has_hue_permission=Mock(return_value=False)),
|
|
|
|
|
+ fs=Mock(
|
|
|
|
|
+ listdir_stats=Mock(return_value=all_stats),
|
|
|
|
|
+ do_as_user=Mock(return_value=all_stats),
|
|
|
|
|
+ isdir=Mock(return_value=True),
|
|
|
|
|
+ normpath=Mock(side_effect=['s3a://test-bucket/test-user/test-dir/file1.txt', 's3a://test-bucket/test-user/test-dir/file2.txt']),
|
|
|
|
|
+ ),
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ mock_paginator = self._create_mock_paginator(all_stats)
|
|
|
|
|
+
|
|
|
|
|
+ with patch('filebrowser.api.Paginator', return_value=mock_paginator) as mock_paginator:
|
|
|
|
|
+ response = listdir_paged(request)
|
|
|
|
|
+ response_data = json.loads(response.content)
|
|
|
|
|
+
|
|
|
|
|
+ assert response.status_code == 200
|
|
|
|
|
+ assert response_data == {
|
|
|
|
|
+ 'files': [
|
|
|
|
|
+ {
|
|
|
|
|
+ 'path': 's3a://test-bucket/test-user/test-dir/file1.txt',
|
|
|
|
|
+ 'aclBit': False,
|
|
|
|
|
+ 'size': 200,
|
|
|
|
|
+ 'atime': 0,
|
|
|
|
|
+ 'mtime': 0,
|
|
|
|
|
+ 'type': 'file',
|
|
|
|
|
+ 'user': 'user2',
|
|
|
|
|
+ 'group': 'group2',
|
|
|
|
|
+ 'mode': 33188,
|
|
|
|
|
+ 'rwx': '-rw-r--r--+',
|
|
|
|
|
+ },
|
|
|
|
|
+ {
|
|
|
|
|
+ 'path': 's3a://test-bucket/test-user/test-dir/file2.txt',
|
|
|
|
|
+ 'aclBit': False,
|
|
|
|
|
+ 'size': 100,
|
|
|
|
|
+ 'atime': 0,
|
|
|
|
|
+ 'mtime': 0,
|
|
|
|
|
+ 'type': 'file',
|
|
|
|
|
+ 'user': 'user1',
|
|
|
|
|
+ 'group': 'group1',
|
|
|
|
|
+ 'mode': 33188,
|
|
|
|
|
+ 'rwx': '-rw-r--r--+',
|
|
|
|
|
+ },
|
|
|
|
|
+ ],
|
|
|
|
|
+ 'page': {'page_number': 1, 'page_size': 30, 'total_pages': 1, 'total_size': 2},
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ # Assert correct sorting
|
|
|
|
|
+ assert response_data['files'][0]['path'] == 's3a://test-bucket/test-user/test-dir/file1.txt'
|
|
|
|
|
+ assert response_data['files'][1]['path'] == 's3a://test-bucket/test-user/test-dir/file2.txt'
|
|
|
|
|
+
|
|
|
|
|
+ def test_listdir_paged_sorting_by_size_descending(self):
|
|
|
|
|
+ with patch('filebrowser.api.is_admin') as is_admin:
|
|
|
|
|
+ is_admin.return_value = False
|
|
|
|
|
+
|
|
|
|
|
+ file1_stats = self._create_mock_file_stats(
|
|
|
|
|
+ name='file1.txt', path='s3a://test-bucket/test-user/test-dir/file1.txt', size=100, user='user1', group='group1'
|
|
|
|
|
+ )
|
|
|
|
|
+ file2_stats = self._create_mock_file_stats(
|
|
|
|
|
+ name='file2.txt', path='s3a://test-bucket/test-user/test-dir/file2.txt', size=200, user='user2', group='group2'
|
|
|
|
|
+ )
|
|
|
|
|
+ file3_stats = self._create_mock_file_stats(
|
|
|
|
|
+ name='file3.txt', path='s3a://test-bucket/test-user/test-dir/file3.txt', size=300, user='user3', group='group3'
|
|
|
|
|
+ )
|
|
|
|
|
+ file4_stats = self._create_mock_file_stats(
|
|
|
|
|
+ name='file4.txt', path='s3a://test-bucket/test-user/test-dir/file4.txt', size=400, user='user4', group='group4'
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ all_stats = [file1_stats, file2_stats, file3_stats, file4_stats]
|
|
|
|
|
+
|
|
|
|
|
+ request = Mock(
|
|
|
|
|
+ method='GET',
|
|
|
|
|
+ GET={
|
|
|
|
|
+ 'pagenum': '1',
|
|
|
|
|
+ 'pagesize': '30',
|
|
|
|
|
+ 'path': 's3a://test-bucket/test-user/test-dir',
|
|
|
|
|
+ 'sortby': 'name',
|
|
|
|
|
+ 'descending': 'True',
|
|
|
|
|
+ },
|
|
|
|
|
+ user=Mock(has_hue_permission=Mock(return_value=False)),
|
|
|
|
|
+ fs=Mock(
|
|
|
|
|
+ listdir_stats=Mock(side_effect=[all_stats, all_stats]),
|
|
|
|
|
+ do_as_user=Mock(return_value=all_stats),
|
|
|
|
|
+ isdir=Mock(return_value=True),
|
|
|
|
|
+ normpath=Mock(
|
|
|
|
|
+ side_effect=[
|
|
|
|
|
+ 's3a://test-bucket/test-user/test-dir/file1.txt',
|
|
|
|
|
+ 's3a://test-bucket/test-user/test-dir/file2.txt',
|
|
|
|
|
+ 's3a://test-bucket/test-user/test-dir/file3.txt',
|
|
|
|
|
+ 's3a://test-bucket/test-user/test-dir/file4.txt',
|
|
|
|
|
+ ]
|
|
|
|
|
+ ),
|
|
|
|
|
+ ),
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ mock_paginator = self._create_mock_paginator(sorted(all_stats, key=lambda x: x.size, reverse=True))
|
|
|
|
|
+
|
|
|
|
|
+ with patch('filebrowser.api.Paginator', return_value=mock_paginator) as mock_paginator:
|
|
|
|
|
+ response = listdir_paged(request)
|
|
|
|
|
+ response_data = json.loads(response.content)
|
|
|
|
|
+
|
|
|
|
|
+ assert response.status_code == 200
|
|
|
|
|
+ assert 'files' in response_data
|
|
|
|
|
+ assert len(response_data['files']) == 4
|
|
|
|
|
+
|
|
|
|
|
+ # Assert correct sorting
|
|
|
|
|
+ assert response_data['files'][0]['size'] == 400
|
|
|
|
|
+ assert response_data['files'][1]['size'] == 300
|
|
|
|
|
+ assert response_data['files'][2]['size'] == 200
|
|
|
|
|
+ assert response_data['files'][3]['size'] == 100
|
|
|
|
|
+
|
|
|
|
|
+ def test_listdir_paged_invalid_path(self):
|
|
|
|
|
+ request = Mock(
|
|
|
|
|
+ method='GET',
|
|
|
|
|
+ GET={'pagenum': '1', 'pagesize': '30', 'path': 's3a://test-bucket/test-user/test-dir/test-file'},
|
|
|
|
|
+ fs=Mock(
|
|
|
|
|
+ isdir=Mock(return_value=False),
|
|
|
|
|
+ ),
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ response = listdir_paged(request)
|
|
|
|
|
+
|
|
|
|
|
+ assert response.status_code == 400
|
|
|
|
|
+ assert response.content.decode('utf-8') == 's3a://test-bucket/test-user/test-dir/test-file is not a directory.'
|
|
|
|
|
+
|
|
|
|
|
+ def test_listdir_paged_error_while_listing_bucket(self):
|
|
|
|
|
+ with patch('filebrowser.api.is_admin') as is_admin:
|
|
|
|
|
+ is_admin.return_value = False
|
|
|
|
|
+
|
|
|
|
|
+ request = Mock(
|
|
|
|
|
+ method='GET',
|
|
|
|
|
+ GET={'pagenum': '1', 'pagesize': '30', 'path': 's3a://test-bucket/'},
|
|
|
|
|
+ fs=Mock(
|
|
|
|
|
+ isdir=Mock(return_value=True),
|
|
|
|
|
+ listdir_stats=Mock(side_effect=S3ListAllBucketsException('Failed to list all buckets')),
|
|
|
|
|
+ do_as_user=Mock(side_effect=S3ListAllBucketsException('Failed to list all buckets')),
|
|
|
|
|
+ ),
|
|
|
|
|
+ )
|
|
|
|
|
+
|
|
|
|
|
+ response = listdir_paged(request)
|
|
|
|
|
+
|
|
|
|
|
+ assert response.status_code == 403
|
|
|
|
|
+ assert response.content.decode('utf-8') == 'Bucket listing is not allowed: Failed to list all buckets'
|