import base64
|
from unittest.mock import MagicMock
|
|
import pytest
|
from django.urls import reverse
|
from io_storages.proxy_api import ProjectResolveStorageUri, TaskResolveStorageUri
|
from projects.models import Project
|
from rest_framework import status
|
from rest_framework.test import APIRequestFactory, force_authenticate
|
from tasks.models import Task
|
from users.models import User
|
|
|
@pytest.mark.django_db
|
class TestTaskResolveStorageUri:
|
@pytest.fixture
|
def view(self):
|
view = TaskResolveStorageUri.as_view()
|
view.authentication_classes = []
|
view.permission_classes = []
|
return view
|
|
@pytest.fixture
|
def project(self):
|
project = Project(pk=1, title='testproject')
|
project.has_permission = MagicMock()
|
project.get_all_import_storage_objects = []
|
return project
|
|
@pytest.fixture
|
def task(self, project):
|
task = Task(pk=1, data={}, project=project)
|
task.resolve_storage_uri = MagicMock()
|
task.has_permission = MagicMock()
|
return task
|
|
@pytest.fixture
|
def user(self):
|
user = User.objects.create_user(username='testuser', email='testuser@email.com', password='testpassword')
|
return user
|
|
def test_missing_parameters(self, view, user):
|
request = APIRequestFactory().get(reverse('storages:task-storage-data-resolve', kwargs={'task_id': 1}))
|
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request)
|
|
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
|
def test_task_not_found(self, view, user):
|
# Test case where task doesn't exist in database
|
request = APIRequestFactory().get(
|
reverse('storages:task-storage-data-resolve', kwargs={'task_id': 2}) + '?fileuri=fileuri'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, task_id=2)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
def test_task_resolve_returns_none(self, view, task, project, user, monkeypatch):
|
# Test case where task exists but resolve_storage_uri returns None
|
task.resolve_storage_uri.return_value = None
|
task.has_permission.return_value = True
|
task.project = project
|
|
def mock_task_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return task
|
else:
|
raise Task.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_task_get
|
monkeypatch.setattr('tasks.models.Task.objects', obj)
|
|
# Add a mock storage that will match the URI
|
mock_storage = MagicMock()
|
mock_storage.presign = True
|
|
def mock_can_resolve(url):
|
return True # Match any URL
|
|
mock_storage.can_resolve_url = MagicMock(side_effect=mock_can_resolve)
|
project.get_all_import_storage_objects = [mock_storage]
|
|
request = APIRequestFactory().get(
|
reverse('storages:task-storage-data-resolve', kwargs={'task_id': 1}) + '?fileuri=fileuri'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, task_id=1)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
def test_storage_resolution_fails(self, view, task, project, user, monkeypatch):
|
task.resolve_storage_uri.return_value = None
|
task.has_permission.return_value = True
|
project.get_all_import_storage_objects = []
|
task.project = project
|
|
def mock_task_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return task
|
else:
|
raise Task.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_task_get
|
monkeypatch.setattr('tasks.models.Task.objects', obj)
|
|
encoded_fileuri = base64.urlsafe_b64encode(b's3://valid/uri').decode()
|
request = APIRequestFactory().get(
|
reverse('storages:task-storage-data-resolve', kwargs={'task_id': 1}) + f'?fileuri={encoded_fileuri}'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, task_id=1)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
task.resolve_storage_uri.assert_not_called()
|
|
def test_file_uri_not_hashed(self, view, task, project, user, monkeypatch):
|
task.has_permission.return_value = True
|
project.get_all_import_storage_objects = []
|
task.project = project
|
|
def mock_task_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return task
|
else:
|
raise Task.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_task_get
|
monkeypatch.setattr('tasks.models.Task.objects', obj)
|
|
request = APIRequestFactory().get(
|
reverse('storages:task-storage-data-resolve', kwargs={'task_id': 1}) + '?fileuri=fileuri'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
|
response = view(request, task_id=1)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
task.resolve_storage_uri.assert_not_called()
|
|
def test_successful_request(self, view, task, project, user, monkeypatch):
|
valid_decoded_uri = "s3://hypertext-bucket/file with /spaces and' / ' / quotes.jpg"
|
encoded_fileuri = base64.urlsafe_b64encode(valid_decoded_uri.encode()).decode()
|
|
mock_storage = MagicMock()
|
mock_storage.presign = True
|
|
def mock_can_resolve(url):
|
return url == valid_decoded_uri
|
|
mock_storage.can_resolve_url = MagicMock(side_effect=mock_can_resolve)
|
|
project.get_all_import_storage_objects = [mock_storage]
|
task.project = project
|
|
task.resolve_storage_uri.return_value = dict(
|
url='https://presigned-url.com/fileuri',
|
presign_ttl=3600,
|
)
|
task.has_permission.return_value = True
|
|
def mock_task_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return task
|
else:
|
raise Task.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_task_get
|
monkeypatch.setattr('tasks.models.Task.objects', obj)
|
|
request = APIRequestFactory().get(
|
reverse('storages:task-storage-data-resolve', kwargs={'task_id': 1}) + f'?fileuri={encoded_fileuri}'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
|
response = view(request, task_id=1)
|
|
assert response.status_code == status.HTTP_303_SEE_OTHER
|
assert response.url == 'https://presigned-url.com/fileuri'
|
mock_storage.can_resolve_url.assert_called_with(valid_decoded_uri)
|
task.resolve_storage_uri.assert_called_once_with(valid_decoded_uri)
|
|
def test_successful_request_with_long_fileuri(self, view, task, project, user, monkeypatch):
|
longest_allowable_cloud_storage_path = 'is/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/long/path/that/needs/to/be/1024/characters.png'
|
longest_uri = f'aaaaa-bbbb://{longest_allowable_cloud_storage_path}'
|
base64_encoded_uri = base64.urlsafe_b64encode(longest_uri.encode()).decode()
|
|
mock_storage = MagicMock()
|
mock_storage.presign = True
|
|
def mock_can_resolve(url):
|
return url == longest_uri
|
|
mock_storage.can_resolve_url = MagicMock(side_effect=mock_can_resolve)
|
|
project.get_all_import_storage_objects = [mock_storage]
|
task.project = project
|
|
task.resolve_storage_uri.return_value = dict(
|
url='https://presigned-url.com/fileuri',
|
presign_ttl=3600,
|
)
|
task.has_permission.return_value = True
|
|
def mock_task_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return task
|
else:
|
raise Task.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_task_get
|
monkeypatch.setattr('tasks.models.Task.objects', obj)
|
|
longest_allowable_url_length = 2000
|
largest_allowable_task_key = 9223372036854775807
|
longest_resolve_path = f'/tasks/{largest_allowable_task_key}/resolve/?fileuri='
|
scheme_length = len('https://')
|
longest_resolve_path_length = len(longest_resolve_path)
|
longest_allowable_fileuri_hash_length = len(base64_encoded_uri)
|
remaining_url_origin_length = (
|
longest_allowable_url_length
|
- scheme_length
|
- longest_resolve_path_length
|
- longest_allowable_fileuri_hash_length
|
)
|
assert remaining_url_origin_length >= 512
|
|
request = APIRequestFactory().get(
|
reverse('storages:task-storage-data-resolve', kwargs={'task_id': 1}) + f'?fileuri={base64_encoded_uri}'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, task_id=1)
|
|
assert response.status_code == status.HTTP_303_SEE_OTHER
|
assert response.url == 'https://presigned-url.com/fileuri'
|
mock_storage.can_resolve_url.assert_called_with(longest_uri)
|
task.resolve_storage_uri.assert_called_once_with(longest_uri)
|
|
|
@pytest.mark.django_db
|
class TestProjectResolveStorageUri:
|
@pytest.fixture
|
def view(self):
|
view = ProjectResolveStorageUri.as_view()
|
view.authentication_classes = []
|
view.permission_classes = []
|
return view
|
|
@pytest.fixture
|
def project(self):
|
project = Project(pk=1, title='testproject')
|
project.resolve_storage_uri = MagicMock()
|
project.has_permission = MagicMock()
|
project.get_all_import_storage_objects = []
|
return project
|
|
@pytest.fixture
|
def user(self):
|
user = User.objects.create_user(username='testuser', email='testuser@email.com', password='testpassword')
|
return user
|
|
def test_missing_parameters(self, view, user):
|
request = APIRequestFactory().get(reverse('storages:project-storage-data-resolve', kwargs={'project_id': 1}))
|
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request)
|
|
assert response.status_code == status.HTTP_400_BAD_REQUEST
|
|
def test_project_not_found(self, view, user):
|
# Test case where project doesn't exist in database
|
request = APIRequestFactory().get(
|
reverse('storages:project-storage-data-resolve', kwargs={'project_id': 2}) + '?fileuri=fileuri'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, project_id=2)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
def test_project_resolve_returns_none(self, view, project, user, monkeypatch):
|
# Test case where project exists but resolve_storage_uri returns None
|
project.resolve_storage_uri.return_value = None
|
project.has_permission.return_value = True
|
|
def mock_project_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return project
|
else:
|
raise Project.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_project_get
|
monkeypatch.setattr('projects.models.Project.objects', obj)
|
|
# Add a mock storage that will match the URI
|
mock_storage = MagicMock()
|
mock_storage.presign = True
|
|
def mock_can_resolve(url):
|
return True # Match any URL
|
|
mock_storage.can_resolve_url = MagicMock(side_effect=mock_can_resolve)
|
project.get_all_import_storage_objects = [mock_storage]
|
|
request = APIRequestFactory().get(
|
reverse('storages:project-storage-data-resolve', kwargs={'project_id': 1}) + '?fileuri=fileuri'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, project_id=1)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
|
def test_storage_resolution_fails(self, view, project, user, monkeypatch):
|
project.resolve_storage_uri.return_value = None
|
project.has_permission.return_value = True
|
|
def mock_project_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return project
|
else:
|
raise Project.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_project_get
|
monkeypatch.setattr('projects.models.Project.objects', obj)
|
|
encoded_fileuri = base64.urlsafe_b64encode(b's3://valid/uri').decode()
|
request = APIRequestFactory().get(
|
reverse('storages:project-storage-data-resolve', kwargs={'project_id': 1}) + f'?fileuri={encoded_fileuri}'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, project_id=1)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
project.resolve_storage_uri.assert_not_called()
|
|
def test_file_uri_not_hashed(self, view, project, user, monkeypatch):
|
project.has_permission.return_value = True
|
|
def mock_project_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return project
|
else:
|
raise Project.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_project_get
|
monkeypatch.setattr('projects.models.Project.objects', obj)
|
|
request = APIRequestFactory().get(
|
reverse('storages:project-storage-data-resolve', kwargs={'project_id': 1}) + '?fileuri=fileuri'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
|
response = view(request, project_id=1)
|
|
assert response.status_code == status.HTTP_404_NOT_FOUND
|
project.resolve_storage_uri.assert_not_called()
|
|
def test_successful_request(self, view, project, user, monkeypatch):
|
valid_decoded_uri = "s3://hypertext-bucket/file with /spaces and' / ' / quotes.jpg"
|
encoded_fileuri = base64.urlsafe_b64encode(valid_decoded_uri.encode()).decode()
|
|
mock_storage = MagicMock()
|
mock_storage.presign = True
|
|
def mock_can_resolve(url):
|
return url == valid_decoded_uri
|
|
mock_storage.can_resolve_url = MagicMock(side_effect=mock_can_resolve)
|
|
project.get_all_import_storage_objects = [mock_storage]
|
|
project.resolve_storage_uri.return_value = dict(
|
url='https://presigned-url.com/fileuri',
|
presign_ttl=3600,
|
)
|
project.has_permission.return_value = True
|
|
def mock_project_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return project
|
else:
|
raise Project.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_project_get
|
monkeypatch.setattr('projects.models.Project.objects', obj)
|
|
request = APIRequestFactory().get(
|
reverse('storages:project-storage-data-resolve', kwargs={'project_id': 1}) + f'?fileuri={encoded_fileuri}'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, project_id=1)
|
|
assert response.status_code == status.HTTP_303_SEE_OTHER
|
assert response.url == 'https://presigned-url.com/fileuri'
|
mock_storage.can_resolve_url.assert_called_with(valid_decoded_uri)
|
project.resolve_storage_uri.assert_called_once_with(valid_decoded_uri)
|
|
def test_successful_request_with_long_fileuri(self, view, project, user, monkeypatch):
|
longest_allowable_cloud_storage_path = 'is/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/a/long/path/that/needs/to/be/1024/characters/long/so/that/it/gets/hashedis/long/path/that/needs/to/be/1024/characters.png'
|
longest_uri = f'aaaaa-bbbb://{longest_allowable_cloud_storage_path}'
|
base64_encoded_uri = base64.urlsafe_b64encode(longest_uri.encode()).decode()
|
|
mock_storage = MagicMock()
|
mock_storage.presign = True
|
|
def mock_can_resolve(url):
|
return url == longest_uri
|
|
mock_storage.can_resolve_url = MagicMock(side_effect=mock_can_resolve)
|
|
project.get_all_import_storage_objects = [mock_storage]
|
|
project.resolve_storage_uri.return_value = dict(
|
url='https://presigned-url.com/fileuri',
|
presign_ttl=3600,
|
)
|
project.has_permission.return_value = True
|
|
def mock_project_get(*args, **kwargs):
|
if kwargs['pk'] == 1:
|
return project
|
else:
|
raise Project.DoesNotExist
|
|
obj = MagicMock()
|
obj.get = mock_project_get
|
monkeypatch.setattr('projects.models.Project.objects', obj)
|
|
longest_allowable_url_length = 2000
|
largest_allowable_project_key = 9223372036854775807
|
longest_resolve_path = f'/projects/{largest_allowable_project_key}/resolve/?fileuri='
|
scheme_length = len('https://')
|
longest_resolve_path_length = len(longest_resolve_path)
|
longest_allowable_fileuri_hash_length = len(base64_encoded_uri)
|
remaining_url_origin_length = (
|
longest_allowable_url_length
|
- scheme_length
|
- longest_resolve_path_length
|
- longest_allowable_fileuri_hash_length
|
)
|
assert remaining_url_origin_length >= 512
|
|
request = APIRequestFactory().get(
|
reverse('storages:project-storage-data-resolve', kwargs={'project_id': 1})
|
+ f'?fileuri={base64_encoded_uri}'
|
)
|
request.user = user
|
force_authenticate(request, user)
|
response = view(request, project_id=1)
|
|
assert response.status_code == status.HTTP_303_SEE_OTHER
|
assert response.url == 'https://presigned-url.com/fileuri'
|
mock_storage.can_resolve_url.assert_called_with(longest_uri)
|
project.resolve_storage_uri.assert_called_once_with(longest_uri)
|
