feat: add signed url for data export download file

futurex_openedx_extensions/helpers/export_csv.py

@@ -8,13 +8,15 @@
 from typing import Any, Generator, Optional, Tuple
 from urllib.parse import urlencode, urlparse
 
+import boto3
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.core.files.base import ContentFile
 from django.core.files.storage import default_storage
 from django.urls import resolve
 from rest_framework.request import Request
 from rest_framework.test import APIRequestFactory
+from storages.backends.s3boto3 import S3Boto3Storage
 
 from futurex_openedx_extensions.helpers.exceptions import FXCodedException, FXExceptionCodes
 from futurex_openedx_extensions.helpers.models import DataExportTask
@@ -240,11 +242,36 @@ def export_data_to_csv(
     )
 
 
+def generate_file_url(storage_path: str) -> str:
+    """
+    Generate a signed URL if default storage is S3, otherwise return the normal URL.
+
+    :param storage_path: The path to the file in storage.
+    :type storage_path: str
+
+    :return: Signed or normal URL.
+    """
+    if not isinstance(default_storage, S3Boto3Storage):
+        return default_storage.url(storage_path)
+
+    s3_client = boto3.client(
+        's3',
+        aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
+        aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
+    )
+    return s3_client.generate_presigned_url(
+        'get_object',
+        Params={'Bucket': settings.AWS_STORAGE_BUCKET_NAME, 'Key': storage_path},
+        HttpMethod='GET',
+        ExpiresIn=3600
+    )
+
+
 def get_exported_file_url(fx_task: DataExportTask) -> Optional[str]:
     """Get file URL"""
     if fx_task.status == fx_task.STATUS_COMPLETED:
         storage_path = os.path.join(_get_storage_dir(str(fx_task.tenant.id)), fx_task.filename)
         if default_storage.exists(storage_path):
-            return default_storage.url(storage_path)
+            return generate_file_url(storage_path)
         log.warning('CSV Export: file not found for completed task %s: %s', fx_task.id, storage_path)
     return None

test_utils/edx_platform_mocks/boto3.py

@@ -0,0 +1,18 @@
+"""Mocked botot3 client """
+
+
+def client(*args, **kwargs):
+    """
+    Fake boto3.client implementation that returns a FakeS3Client.
+    Mimics the behavior of the real boto3.client but without making any actual API calls.
+    """
+    class FakeS3Client:  # pylint: disable=too-few-public-methods
+        """
+        A fake S3 client to mock boto3 client behavior.
+        """
+        def generate_presigned_url(self, *args, **kwargs):  # pylint: disable=no-self-use
+            params = kwargs.get('Params', {})
+            file_key = params.get('Key', 'default-file.csv')
+            return f'http://fake-s3-url.com/signed-{file_key}'
+
+    return FakeS3Client()

test_utils/edx_platform_mocks/storages/backends/s3boto3.py

@@ -0,0 +1,9 @@
+"""Fake S3Boto3Storage"""
+
+
+class S3Boto3Storage:  # pylint: disable=too-few-public-methods
+    """
+    A minimal class to simulate the behavior of S3Boto3Storage for mocking purposes.
+    This class does not implement any actual functionality but can only be used in tests.
+    """
+    pass  # pylint: disable=unnecessary-pass

tests/test_helpers/test_export_csv.py

@@ -22,6 +22,7 @@
     _paginated_response_generator,
     _upload_file_to_storage,
     export_data_to_csv,
+    generate_file_url,
     get_exported_file_url,
 )
 from futurex_openedx_extensions.helpers.models import DataExportTask
@@ -465,12 +466,14 @@ def test_export_data_to_csv_for_filename_extension(
     (False, True, None)
 ])
 @patch('futurex_openedx_extensions.helpers.export_csv.default_storage')
+@patch('futurex_openedx_extensions.helpers.export_csv.generate_file_url')
 def test_get_exported_file_url(
-    mock_storage, is_file_exist, is_task_completed, expected_return_value, user
-):  # pylint: disable=redefined-outer-name
+    mocked_generate_url, mock_storage, is_file_exist, is_task_completed, expected_return_value, user
+):  # pylint: disable=redefined-outer-name, too-many-arguments
     """Test get exported file URL"""
     mock_storage.exists.return_value = is_file_exist
     mock_storage.url.return_value = 'http://example.com/exported_file.csv' if is_file_exist else None
+    mocked_generate_url.return_value = expected_return_value
     task = DataExportTask.objects.create(
         tenant_id=1,
         filename='exported_file.csv',
@@ -487,3 +490,60 @@ def test_export_data_to_csv_for_url(fx_task):  # pylint: disable=redefined-outer
     with pytest.raises(FXCodedException) as exc_info:
         export_data_to_csv(fx_task.id, url_with_query_str, {}, {}, 'test.csv')
     assert str(exc_info.value) == f'CSV Export: Unable to process URL with query params: {url_with_query_str}'
+
+
+@override_settings(
+    AWS_STORAGE_BUCKET_NAME='fake-bucket',
+    AWS_ACCESS_KEY_ID='fake-id',
+    AWS_SECRET_ACCESS_KEY='fake-access-key'
+)
+@pytest.mark.parametrize('is_s3_instance, expected_return_value, usecase', [
+    (True, 'http://fake-s3-url.com/signed-fake-path', 'S3 storage use case failed'),
+    (False, 'http://default_storage_url/fake-path', 'Other storage use case failed'),
+])
+@patch('futurex_openedx_extensions.helpers.export_csv.default_storage')
+@patch('futurex_openedx_extensions.helpers.export_csv.isinstance')
+def test_generate_file_url_for_return_value(
+    mocked_is_instance, mocked_default_storage, is_s3_instance, expected_return_value, usecase
+):
+    """
+    test the behavior of `generate_file_url` function
+    for both S3-based storage and non-S3 storage backends.
+
+    Test cases:
+    1. When the default storage is an instance of S3Boto3Storage:
+       - The function should return the signed URL generated by `generate_presigned_url`.
+    2. When the default storage is not an instance of S3Boto3Storage:
+       - The function should return the URL provided by the `default_storage.url()` method.
+    """
+    dummy_file_path = 'fake-path'
+    mocked_is_instance.return_value = is_s3_instance
+    mocked_default_storage.url.return_value = 'http://default_storage_url/fake-path'
+    assert generate_file_url(dummy_file_path) == expected_return_value, usecase
+
+
+@override_settings(
+    AWS_STORAGE_BUCKET_NAME='fake-bucket',
+    AWS_ACCESS_KEY_ID='fake-id',
+    AWS_SECRET_ACCESS_KEY='fake-access-key'
+)
+@patch('futurex_openedx_extensions.helpers.export_csv.isinstance', return_value=True)
+@patch('boto3.client')
+def test_generate_file_url_for_s3_storage(
+    mocked_boto3_client, mocked_is_instance
+):  # pylint: disable=unused-argument
+    """
+    test generate_file_url funtionality for s3 storage that
+    needed s3 storage fucntions are called with correct arguments
+    """
+    dummy_file_path = 'fake/file.csv'
+    generate_file_url(dummy_file_path)
+    mocked_boto3_client.assert_called_once_with(
+        's3', aws_access_key_id='fake-id', aws_secret_access_key='fake-access-key'
+    )
+    mocked_boto3_client.return_value.generate_presigned_url.assert_called_once_with(
+        'get_object',
+        Params={'Bucket': 'fake-bucket', 'Key': dummy_file_path},
+        HttpMethod='GET',
+        ExpiresIn=3600
+    )