Pilot 3546: add folder merging feature

app/resources/custom_error.py

@@ -53,6 +53,8 @@ class Error:
         ),
         'UPLOAD_CANCEL': 'Upload task was cancelled.',
         'UPLOAD_FAIL': 'Upload task was failed. Please check the console output.',
+        'UPLOAD_SKIP_DUPLICATION': 'Following files with the same '
+        'name already exist in the Project: \n%s.\nDo you want to skip uploading',
         'UPLOAD_ID_NOT_EXIST': (
             'The specified multipart upload does not exist. '
             'The upload ID may be invalid, or the upload may have been aborted or completed.'

app/services/file_manager/file_upload/file_upload.py

@@ -2,7 +2,6 @@
 #
 # Contact Indoc Systems for any questions regarding the use of this source code.
 
-import math
 import os
 import time
 import zipfile
@@ -24,6 +23,7 @@
 from app.services.output_manager.error_handler import ECustomizedError
 from app.services.output_manager.error_handler import SrvErrorHandler
 from app.services.output_manager.error_handler import customized_error_msg
+from app.utils.aggregated import batch_generator
 from app.utils.aggregated import get_file_in_folder
 from app.utils.aggregated import get_file_info_by_geid
 from app.utils.aggregated import search_item
@@ -174,18 +174,32 @@ def simple_upload(  # noqa: C901
         if file_object.total_size == 0:
             logger.warning(f'Skip the file with 0 size: {file_object.file_name}')
         else:
-            file_objects.append(FileObject(object_path, file))
+            file_objects.append(file_object)
+
+    # make the file duplication check to allow folde merging
+    non_duplicate_file_objects = []
+    if create_folder_flag is True:
+        non_duplicate_file_objects = file_objects
+    else:
+        mhandler.SrvOutPutHandler.file_duplication_check()
+        duplicated_file = []
+        for file_batchs in batch_generator(file_objects, batch_size=AppConfig.Env.upload_batch_size):
+            non_duplicates, duplicate_path = upload_client.check_upload_duplication(file_batchs)
+            non_duplicate_file_objects.extend(non_duplicates)
+            duplicated_file.extend(duplicate_path)
+
+        if len(non_duplicate_file_objects) == 0:
+            mhandler.SrvOutPutHandler.file_duplication_check_warning_with_all_same()
+        elif len(duplicated_file) > 0:
+            mhandler.SrvOutPutHandler.file_duplication_check_success()
+            duplicate_warning_format = '\n'.join(duplicated_file)
+            click.confirm(
+                customized_error_msg(ECustomizedError.UPLOAD_SKIP_DUPLICATION) % (duplicate_warning_format), abort=True
+            )
 
-    # here add the batch of 500 per loop, the pre upload api cannot
-    # process very large amount of file at same time. otherwise it will timeout
-    num_of_batchs = math.ceil(len(file_objects) / AppConfig.Env.upload_batch_size)
     # here is list of pre upload result. We decided to call pre upload api by batch
     pre_upload_infos = []
-    for batch in range(0, num_of_batchs):
-        start_index = batch * AppConfig.Env.upload_batch_size
-        end_index = (batch + 1) * AppConfig.Env.upload_batch_size
-        file_batchs = file_objects[start_index:end_index]
-
+    for file_batchs in batch_generator(non_duplicate_file_objects, batch_size=AppConfig.Env.upload_batch_size):
         # sending the pre upload request to generate
         # the placeholder in object storage
         pre_upload_infos.extend(upload_client.pre_upload(file_batchs, output_path))
@@ -266,12 +280,8 @@ def resume_upload(
 
     # here add the batch of 500 per loop, the pre upload api cannot
     # process very large amount of file at same time. otherwise it will timeout
-    num_of_batchs = math.ceil(len(all_files) / AppConfig.Env.upload_batch_size)
     # here is list of pre upload result. We decided to call pre upload api by batch
-    for batch in range(0, num_of_batchs):
-        start_index = batch * AppConfig.Env.upload_batch_size
-        end_index = (batch + 1) * AppConfig.Env.upload_batch_size
-        file_batchs = item_ids[start_index:end_index]
+    for file_batchs in batch_generator(item_ids, batch_size=AppConfig.Env.upload_batch_size):
         items = get_file_info_by_geid(file_batchs)
 
         # get the detail of item to see if the file is already uploaded

app/services/file_manager/file_upload/upload_client.py

@@ -147,6 +147,40 @@ def resume_upload(self, unfinished_file_objects: List[FileObject]) -> List[FileO
 
         return unfinished_file_objects
 
+    @require_valid_token()
+    def check_upload_duplication(self, file_objects: List[FileObject]) -> Tuple[List[FileObject], List[str]]:
+        """
+        Summary:
+            The function will call the api to check if the file has been uploaded.
+            if yes, it will skip the file.
+        Parameter:
+            - file_objects(List[FileObject]): the file will be uploaded.
+        return:
+            - non_exist_file_objects(List[FileObject]): the file that need to be uploaded.
+            - exist_files(List[str]): the file that has been uploaded. will be skipped
+        """
+        headers = {'Authorization': 'Bearer ' + self.user.access_token, 'Session-ID': self.user.session_id}
+        url = AppConfig.Connections.url_base + '/portal/v1/files/exists'
+
+        # generate a list of locations for uploaded files to check duplication
+        # at same time, generate a dict of mapping with object_path: FileObject
+        locations = [x.object_path for x in file_objects]
+        object_path_file_object_map = {x.object_path: x for x in file_objects}
+
+        payload = {'locations': locations, 'container_code': self.project_code, 'container_type': 'project', 'zone': 0}
+        response = resilient_session().post(url, json=payload, headers=headers)
+
+        # pop the file object if the file has been uploaded
+        # return the file objects that need to be uploaded
+        if response.status_code == 200:
+            exist_files = response.json().get('result', [])
+            for exist_file_path in exist_files:
+                object_path_file_object_map.pop(exist_file_path)
+        else:
+            SrvErrorHandler.default_handle('Error when checking file duplication', if_exit=True)
+
+        return list(object_path_file_object_map.values()), exist_files
+
     @require_valid_token()
     def pre_upload(self, file_objects: List[FileObject], output_path: str) -> List[FileObject]:
         """
@@ -179,7 +213,7 @@ def pre_upload(self, file_objects: List[FileObject], output_path: str) -> List[F
             ],
         }
 
-        response = resilient_session().post(url, json=payload, headers=headers, timeout=None)
+        response = resilient_session().post(url, json=payload, headers=headers)
         if response.status_code == 200:
             result = response.json().get('result')
             file_mapping = {x.object_path: x for x in file_objects}

app/services/output_manager/error_handler.py

@@ -37,6 +37,7 @@ class ECustomizedError(enum.Enum):
     PERMISSION_DENIED = 'PERMISSION_DENIED'
     UPLOAD_CANCEL = 'UPLOAD_CANCEL'
     UPLOAD_FAIL = 'UPLOAD_FAIL'
+    UPLOAD_SKIP_DUPLICATION = 'UPLOAD_SKIP_DUPLICATION'
     # the error when multipart upload id is not exist
     UPLOAD_ID_NOT_EXIST = 'UPLOAD_ID_NOT_EXIST'
     MANIFEST_OF_FOLDER_FILE_EXIST = 'MANIFEST_OF_FOLDER_FILE_EXIST'

app/services/output_manager/message_handler.py

@@ -141,6 +141,21 @@ def preupload_success():
         """e.g. pre-upload succeed."""
         return logger.info('Pre-upload complete.')
 
+    @staticmethod
+    def file_duplication_check():
+        """e.g. file duplication check."""
+        return logger.info('Checking for file duplication...')
+
+    @staticmethod
+    def file_duplication_check_success():
+        """e.g. file duplication check succeed."""
+        return logger.info('File duplication check complete.')
+
+    @staticmethod
+    def file_duplication_check_warning_with_all_same():
+        """e.g. file duplication check warning with all same."""
+        return logger.warning('All files are the same, no need to upload.')
+
     @staticmethod
     def resume_check_success():
         """e.g. notify the resumable check succeed."""

app/utils/aggregated.py

@@ -5,6 +5,8 @@
 import os
 import re
 import shutil
+from typing import Any
+from typing import List
 
 import httpx
 import requests
@@ -132,3 +134,9 @@ def identify_target_folder(project_path):
         SrvErrorHandler.customized_handle(ECustomizedError.INVALID_NAMEFOLDER, True)
         target_folder = ''
     return project_code, target_folder
+
+
+def batch_generator(iterable: List[Any], batch_size=1):
+    max_size = len(iterable)
+    for start_index in range(0, max_size, batch_size):
+        yield iterable[start_index : min(start_index + batch_size, max_size)]

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "app"
-version = "2.7.0a0"
+version = "2.7.0"
 description = "This service is designed to support pilot platform"
 authors = ["Indoc Systems"]
 

tests/app/services/file_manager/file_upload/test_file_upload.py

@@ -174,6 +174,87 @@ def test_dont_allow_attribute_attaching_when_folder_upload(mocker, capfd):
         AssertionError('SystemExit not raised')
 
 
+def test_folder_merge_succuss_with_no_duplication(mocker, mock_upload_client):
+    file_name = 'test'
+    upload_event = {
+        'file': file_name,
+        'project_code': 'test_project',
+        'zone': 'greenroom',
+        'create_folder_flag': False,
+    }
+
+    mocker.patch('os.path.isdir', return_value=False)
+    mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1))
+
+    non_dup_list = [FileObject('object/path', 'local_path', 'resumable_id', 'job_id', 'item_id')]
+    mocker.patch(
+        'app.services.file_manager.file_upload.file_upload.UploadClient.check_upload_duplication',
+        return_value=(non_dup_list, []),
+    )
+
+    item_ids = simple_upload(upload_event)
+    assert len(item_ids) == 1
+    assert item_ids[0] == non_dup_list[0].item_id
+
+
+def test_folder_merge_succuss_with_duplication(mocker, mock_upload_client):
+    file_name = 'test'
+    upload_event = {
+        'file': file_name,
+        'project_code': 'test_project',
+        'zone': 'greenroom',
+        'create_folder_flag': False,
+    }
+
+    mocker.patch('os.path.isdir', return_value=False)
+    mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1))
+    click_yes_mock = mocker.patch('app.services.file_manager.file_upload.file_upload.click.confirm', return_value=None)
+
+    non_dup_list = [FileObject('object/path', 'local_path', 'resumable_id', 'job_id', 'item_id')]
+    dup_list = ['object/dup']
+    mocker.patch(
+        'app.services.file_manager.file_upload.file_upload.UploadClient.check_upload_duplication',
+        return_value=(non_dup_list, dup_list),
+    )
+
+    item_ids = simple_upload(upload_event)
+    assert len(item_ids) == 1
+    assert item_ids[0] == non_dup_list[0].item_id
+    assert click_yes_mock.call_count == 1
+
+
+def test_folder_merge_skip_with_all_duplication(mocker, mock_upload_client, capfd):
+    file_name = 'test'
+    upload_event = {
+        'file': file_name,
+        'project_code': 'test_project',
+        'zone': 'greenroom',
+        'create_folder_flag': False,
+    }
+
+    mocker.patch('os.path.isdir', return_value=False)
+    mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1))
+    click_yes_mock = mocker.patch('app.services.file_manager.file_upload.file_upload.click.confirm', return_value=None)
+
+    dup_list = ['object/dup']
+    mocker.patch(
+        'app.services.file_manager.file_upload.file_upload.UploadClient.check_upload_duplication',
+        return_value=([], dup_list),
+    )
+
+    item_ids = simple_upload(upload_event)
+    assert len(item_ids) == 0
+    assert click_yes_mock.call_count == 0
+
+    out, _ = capfd.readouterr()
+    expect = (
+        f'Starting upload of: {file_name}\n'
+        + 'Checking for file duplication...\n'
+        + 'All files are the same, no need to upload.\n'
+    )
+    assert expect in out
+
+
 def test_resume_upload(mocker):
     mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1))
     test_obj = FileObject('object/path', 'local_path', 'resumable_id', 'job_id', 'item_id')

tests/app/services/file_manager/file_upload/test_upload_client.py

@@ -153,6 +153,54 @@ def test_resumable_pre_upload_failed_with_404(httpx_mock, mocker):
         AssertionError('SystemExit not raised')
 
 
+def test_check_upload_duplication_success(httpx_mock, mocker):
+    mocker.patch(
+        'app.services.user_authentication.token_manager.SrvTokenManager.decode_access_token',
+        return_value=decoded_token(),
+    )
+    upload_client = UploadClient('project_code', 'parent_folder_id')
+    mocker.patch('app.services.file_manager.file_upload.models.FileObject.generate_meta', return_value=(1, 1))
+    dup_obj = FileObject('object/duplicate', 'local_path', 'resumable_id', 'job_id', 'item_id')
+    not_dup_object = FileObject('object/not_duplicate', 'local_path', 'resumable_id', 'job_id', 'item_id')
+
+    url = AppConfig.Connections.url_base + '/portal/v1/files/exists'
+    httpx_mock.add_response(
+        method='POST',
+        url=url,
+        json={'result': [dup_obj.object_path]},
+    )
+
+    not_dup_list, dup_list = upload_client.check_upload_duplication([dup_obj, not_dup_object])
+    assert not_dup_list == [not_dup_object]
+    assert dup_list == [dup_obj.object_path]
+
+
+def test_check_upload_duplication_fail_with_500(httpx_mock, mocker, capfd):
+    mocker.patch(
+        'app.services.user_authentication.token_manager.SrvTokenManager.decode_access_token',
+        return_value=decoded_token(),
+    )
+    upload_client = UploadClient('project_code', 'parent_folder_id')
+
+    url = AppConfig.Connections.url_base + '/portal/v1/files/exists'
+    httpx_mock.add_response(
+        method='POST',
+        url=url,
+        json={'result': []},
+        status_code=500,
+    )
+
+    try:
+        upload_client.check_upload_duplication([])
+    except SystemExit:
+        out, _ = capfd.readouterr()
+
+        expect = 'Error when checking file duplication\n'
+        assert out == expect
+    else:
+        AssertionError('SystemExit not raised')
+
+
 def test_output_manifest_success(mocker, tmp_path):
     upload_client = UploadClient('project_code', 'parent_folder_id')
     json_dump_mocker = mocker.patch('json.dump', return_value=None)

tests/conftest.py

@@ -41,12 +41,24 @@ def user_login_true(mocker):
     mocker.patch('app.services.user_authentication.decorator.check_is_active', return_value=True)
 
 
+@pytest.fixture
+def mock_upload_client(monkeypatch):
+    from app.services.file_manager.file_upload.upload_client import UploadClient
+
+    monkeypatch.setattr(UploadClient, 'pre_upload', lambda *args, **kwargs: args[1])
+    monkeypatch.setattr(UploadClient, 'stream_upload', lambda *args, **kwargs: [])
+    monkeypatch.setattr(UploadClient, 'on_succeed', lambda *args, **kwargs: None)
+    monkeypatch.setattr(UploadClient, 'output_manifest', lambda *args, **kwargs: {})
+    monkeypatch.setattr(UploadClient, 'check_status', lambda *args, **kwargs: True)
+
+
 @pytest.fixture
 def settings() -> Settings:
     return get_settings()
 
 
 def decoded_token():
+    setting = get_settings()
     current_time = int(time.time()) + 1000
     return {
         'exp': current_time + 100,
@@ -57,7 +69,7 @@ def decoded_token():
         'aud': 'account',
         'sub': 'a8b728f6-c95a-4999-b98e-0ccf7492a9b4',
         'typ': 'Bearer',
-        'azp': 'cli',
+        'azp': setting.keycloak_device_client_id,
         'nonce': 'a3cb03d0-b00a-480d-8fd2-e06f80898cf1',
         'session_state': 'b92a3847-a485-4060-91fd-83300b09acb6',
         'acr': '1',