diff --git a/app/resources/custom_error.py b/app/resources/custom_error.py index 253d8b9c..e708703e 100644 --- a/app/resources/custom_error.py +++ b/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.' diff --git a/app/services/file_manager/file_upload/file_upload.py b/app/services/file_manager/file_upload/file_upload.py index a9000e6d..aa8ae001 100644 --- a/app/services/file_manager/file_upload/file_upload.py +++ b/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 diff --git a/app/services/file_manager/file_upload/upload_client.py b/app/services/file_manager/file_upload/upload_client.py index 202986f5..cf3fb7a7 100644 --- a/app/services/file_manager/file_upload/upload_client.py +++ b/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} diff --git a/app/services/output_manager/error_handler.py b/app/services/output_manager/error_handler.py index 1902d0f7..d855013f 100644 --- a/app/services/output_manager/error_handler.py +++ b/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' diff --git a/app/services/output_manager/message_handler.py b/app/services/output_manager/message_handler.py index 0afb6de6..d5b179ac 100644 --- a/app/services/output_manager/message_handler.py +++ b/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.""" diff --git a/app/utils/aggregated.py b/app/utils/aggregated.py index 440e26e5..725cb753 100644 --- a/app/utils/aggregated.py +++ b/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)] diff --git a/pyproject.toml b/pyproject.toml index 40a0f3e6..1ac9ae0c 100644 --- a/pyproject.toml +++ b/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"] diff --git a/tests/app/services/file_manager/file_upload/test_file_upload.py b/tests/app/services/file_manager/file_upload/test_file_upload.py index b3b4d9b6..fcc1c726 100644 --- a/tests/app/services/file_manager/file_upload/test_file_upload.py +++ b/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') diff --git a/tests/app/services/file_manager/file_upload/test_upload_client.py b/tests/app/services/file_manager/file_upload/test_upload_client.py index 93f2b036..d1809d0b 100644 --- a/tests/app/services/file_manager/file_upload/test_upload_client.py +++ b/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) diff --git a/tests/conftest.py b/tests/conftest.py index 8d85ac44..8c74fa46 100644 --- a/tests/conftest.py +++ b/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',