feat: add signed url for data export download file

futurex_openedx_extensions/helpers/export_csv.py

@@ -1,12 +1,14 @@
 """
 This module contains utils for tasks.
 """
+import boto3
 import csv
 import logging
 import os
 import tempfile
 from typing import Any, Generator, Optional, Tuple
 from urllib.parse import urlencode, urlparse
+from storages.backends.s3boto3 import S3Boto3Storage
 
 from django.conf import settings
 from django.contrib.auth import get_user_model
@@ -240,11 +242,36 @@ def export_data_to_csv(
     )
 
 
+def generate_file_url(storage_path: 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,21 @@
+"""Mocked botot3 client """
+
+
+def client(service_name, *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:
+        """
+        A fake S3 client to mock boto3 client behavior.
+        """
+        def generate_presigned_url(self, *args, **kwargs):
+            params = kwargs.get('Params', {})
+            file_key = params.get('Key', 'default-file.csv')
+            return f'http://fake-s3-url.com/signed-{file_key}'
+
+    if service_name == 's3':
+        return FakeS3Client()
+    else:
+        raise ValueError(f"Service {service_name} not supported in fake client.")

test_utils/edx_platform_mocks/storages/backends/s3boto3.py

@@ -0,0 +1,9 @@
+"""Fake S3Boto3Storage"""
+
+
+class S3Boto3Storage:
+    """
+    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 

tests/test_helpers/test_export_csv.py

@@ -23,8 +23,11 @@
     _upload_file_to_storage,
     export_data_to_csv,
     get_exported_file_url,
+    generate_file_url
 )
 from futurex_openedx_extensions.helpers.models import DataExportTask
+from storages.backends.s3boto3 import S3Boto3Storage
+
 
 _FILENAME = 'test.csv'
 
@@ -465,12 +468,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
+    mocked_generate_url, mock_storage, is_file_exist, is_task_completed, expected_return_value, user
 ):  # pylint: disable=redefined-outer-name
     """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 +492,40 @@ 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}'
+
+
+@patch('futurex_openedx_extensions.helpers.export_csv.default_storage')
+def test_generate_file_url_for_non_s3(mocked_default_storage):
+    """
+    test generate_file_url funtionality
+    """
+    dummy_file_path = 'fake/file.csv'
+    dummy_url = 'http://example.com/exported_file.csv'
+    mocked_default_storage.return_value = MagicMock()
+    mocked_default_storage.url.return_value = dummy_url
+    generate_file_url(dummy_file_path) == dummy_url
+    mocked_default_storage.url.assert_called_once_with(dummy_file_path)
+
+
+@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_signed_url(mocked_boto3_client, mocked_is_instance):
+    """
+    test generate_file_url funtionality
+    """
+    dummy_file_path = 'fake/file.csv'
+    generate_file_url(dummy_file_path) == 'http://fake-s3-url.com/signed-{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
+    )