Pilot 4730: update upload command to distinguish project folder and name folder

app/commands/file.py

@@ -167,8 +167,7 @@ def file_put(**kwargs):  # noqa: C901
         message_handler.SrvOutPutHandler.cancel_upload()
         exit(1)
 
-    project_path = click.prompt('ProjectCode') if not project_path else project_path
-    project_code, target_folder = identify_target_folder(project_path)
+    project_code, folder_type, target_folder = identify_target_folder(project_path)
     srv_manifest = SrvFileManifests()
     upload_val_event = {
         'zone': zone,
@@ -207,8 +206,8 @@ def file_put(**kwargs):  # noqa: C901
             f,
             target_folder,
             project_code,
+            folder_type,
             zone,
-            zipping,
         )
 
         upload_event = {

app/models/folder.py

@@ -0,0 +1,22 @@
+# Copyright (C) 2023-2024 Indoc Systems
+#
+# Contact Indoc Systems for any questions regarding the use of this source code.
+
+from enum import Enum
+
+
+class FolderType(str, Enum):
+    """Available folder types."""
+
+    NAMEFOLDER = 'namefolder'
+    PROJECTFOLDER = 'projectfolder'
+
+    def get_prefix(self) -> str:
+        """Get the prefix for the folder type."""
+
+        prefix = {
+            'namefolder': '',
+            'projectfolder': 'shared/',
+        }
+
+        return prefix.get(self.value)

app/resources/custom_help.py

@@ -43,7 +43,11 @@ class HelpPage:
             'FILE_SYNC_ZIP': 'Download files as a zip.',
             'FILE_SYNC_I': 'Enable downloading by geid.',
             'FILE_SYNC_Z': 'Target Zone (i.e., core/greenroom).',
-            'FILE_UPLOAD_P': 'Project folder path starting from Project Code. (i.e., indoctestproject/user/folder)',
+            'FILE_UPLOAD_P': (
+                'Project folder path starting from Project Code(i.e. <project>/user/folder). '
+                'A new key word `projectfolder` is required to specify project folder(i.e. '
+                '<project>/projectfolder/folder1)'
+            ),
             'FILE_UPLOAD_A': 'Add attributes to the file using a File Attribute Template.',
             'FILE_UPLOAD_T': 'Add tags to the file using a Tag file.',
             'FILE_UPLOAD_M': 'The message used to comment on the purpose of uploading your processed file.',

app/services/file_manager/file_upload/file_upload.py

@@ -18,6 +18,7 @@
 import app.services.logger_services.log_functions as logger
 import app.services.output_manager.message_handler as mhandler
 from app.configs.app_config import AppConfig
+from app.models.folder import FolderType
 from app.services.file_manager.file_upload.models import FileObject
 from app.services.file_manager.file_upload.models import ItemStatus
 from app.services.file_manager.file_upload.models import UploadType
@@ -43,7 +44,7 @@ def compress_folder_to_zip(path):
 
 
 def assemble_path(
-    f: str, target_folder: str, project_code: str, zone: str, zipping: bool = False
+    f: str, target_folder: str, project_code: str, folder_type: FolderType, zone: str
 ) -> Tuple[str, Dict, bool, str]:
     '''
     Summary:
@@ -61,7 +62,6 @@ def assemble_path(
          - target_folder(str): the folder on the platform
          - project_code(str): the unique identifier of project
          - zone(str): the zone label eg.greenroom/core
-         - zipping(bool): default False. The flag to indicate if upload as a zip
     Return:
          - current_file_path: the format file path on platform
          - parent_folder: the item information of longest parent folder
@@ -72,18 +72,16 @@ def assemble_path(
 
     current_file_path = target_folder + '/' + f.rstrip('/').split('/')[-1]
     # set name folder as first parent folder
-    name_folder = target_folder.split('/')[0]
-    parent_folder = search_item(project_code, zone, name_folder).get('result', {})
+    root_folder = target_folder.split('/')[0]
+    parent_folder = search_item(project_code, zone, root_folder).get('result', {})
 
     # if f input is a file then current_folder_node is target_folder
     # otherwise it is target_folder + f input name
     current_folder_node = target_folder if os.path.isfile(f) else current_file_path
     create_folder_flag = False
-    # always add `shared/` as prefix to folder/file if
-    # they directly under the project root folder
-    if parent_folder.get('type') == 'project_folder':
-        current_folder_node = 'shared/' + current_folder_node
-        target_folder = 'shared/' + target_folder
+    # add prefix to folder
+    current_folder_node = folder_type.get_prefix() + current_folder_node
+    target_folder = folder_type.get_prefix() + target_folder
 
     if len(current_file_path.split('/')) > 2:
         sub_path = target_folder.split('/')

app/utils/aggregated.py

@@ -8,6 +8,7 @@
 from typing import Any
 from typing import Dict
 from typing import List
+from typing import Tuple
 
 import httpx
 import requests
@@ -16,6 +17,7 @@
 from app.configs.app_config import AppConfig
 from app.configs.config import ConfigClass
 from app.configs.user_config import UserConfig
+from app.models.folder import FolderType
 from app.services.output_manager.error_handler import ECustomizedError
 from app.services.output_manager.error_handler import SrvErrorHandler
 from app.services.user_authentication.decorator import require_valid_token
@@ -164,19 +166,49 @@ def get_file_in_folder(path):
     return files_list
 
 
-def identify_target_folder(project_path):
-    project_code = project_path.split('/')[0]
-    if len(project_path.split('/')) > 1:
-        target_folder = '/'.join(project_path.split('/')[1:])
-        for f in target_folder.split('/'):
-            f = f.strip(' ')
-            valid = validate_folder_name(f)
-            if not valid:
-                SrvErrorHandler.customized_handle(ECustomizedError.INVALID_FOLDERNAME, True)
+def identify_target_folder(project_path: str) -> Tuple[str, FolderType, str]:
+    '''
+    Summary:
+        the function will validate if input folder path doesn't
+        contain invalid characters and return the project code and target folder
+    Parameters:
+        - project_path:
+            - for project folder the input folder path (eg. <project_code>/projectfolder/<folder_name>)
+            - for name folder the input folder path will be (eg. <project_code>/<folder_name>)
+    Return:
+        - project_code: the project code
+        - folder_type: the folder type
+        - target_folder: the target folder
+    '''
+    # split into project_code, folder_type, folder
+    temp_paths = project_path.split('/', 2)
+    project_code, folder_type, folder_name = temp_paths[0], '', ''
+
+    # check folder type if is project folder or name folder
+    # there will be a extra string for project folder between project code and folder name
+    if len(temp_paths) == 2:
+        folder_type = FolderType.NAMEFOLDER
+        folder_name = temp_paths[1]
+    elif len(temp_paths) >= 3:
+        if temp_paths[1] == FolderType.PROJECTFOLDER.value:
+            folder_type = FolderType.PROJECTFOLDER
+            folder_name = temp_paths[2]
+        else:
+            folder_type = FolderType.NAMEFOLDER
+            folder_name = os.path.join(temp_paths[1], temp_paths[2])
     else:
         SrvErrorHandler.customized_handle(ECustomizedError.INVALID_NAMEFOLDER, True)
         target_folder = ''
-    return project_code, target_folder
+
+    # first check if folder names are valid
+    target_folder = '/'.join(folder_name.split('/'))
+    for f in target_folder.split('/'):
+        f = f.strip(' ')
+        valid = validate_folder_name(f)
+        if not valid:
+            SrvErrorHandler.customized_handle(ECustomizedError.INVALID_FOLDERNAME, True)
+
+    return project_code, folder_type, target_folder
 
 
 def batch_generator(iterable: List[Any], batch_size=1):

pyproject.toml

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

tests/app/commands/test_file.py

@@ -24,10 +24,6 @@
 
 
 def test_file_upload_command_success_with_attribute(mocker, cli_runner):
-    project_code = 'test_project'
-    target_folder = 'admin'
-
-    mocker.patch('app.commands.file.identify_target_folder', return_value=(project_code, target_folder))
     mocker.patch('app.commands.file.validate_upload_event', return_value={'source_file': '', 'attribute': 'test'})
     mocker.patch('app.commands.file.assemble_path', return_value=('test', {'id': 'id'}, True, 'test'))
 
@@ -48,8 +44,10 @@ def test_file_upload_command_success_with_attribute(mocker, cli_runner):
             json.dump({'template': {'attr1': 'value'}}, f)
 
         result = cli_runner.invoke(
-            file_put, ['--project-path', 'test', '--thread', 1, '--attribute', 'template.json', 'test.txt']
+            file_put,
+            ['--project-path', 'test_project/admin', '--thread', 1, '--attribute', 'template.json', 'test.txt'],
         )
+
     assert result.exit_code == 0
     simple_upload_mock.assert_called_once()
     attribute_mock.assert_called_once()

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

@@ -3,6 +3,7 @@
 # Contact Indoc Systems for any questions regarding the use of this source code.
 
 from app.configs.app_config import AppConfig
+from app.models.folder import FolderType
 from app.services.file_manager.file_upload.file_upload import assemble_path
 from app.services.file_manager.file_upload.file_upload import resume_upload
 from app.services.file_manager.file_upload.file_upload import simple_upload
@@ -17,7 +18,6 @@ def test_assemble_path_at_name_folder(mocker):
     target_folder = 'admin'
     project_code = 'test_project'
     zone = 0
-    resumable_id = None
 
     mocker.patch(
         'app.services.file_manager.file_upload.file_upload.search_item',
@@ -34,7 +34,7 @@ def test_assemble_path_at_name_folder(mocker):
     )
 
     current_file_path, parent_folder, create_folder_flag, _ = assemble_path(
-        local_file_path, target_folder, project_code, zone, resumable_id
+        local_file_path, target_folder, project_code, FolderType.NAMEFOLDER, zone
     )
     assert current_file_path == 'admin/file.txt'
     assert parent_folder.get('name') == 'admin'
@@ -46,7 +46,6 @@ def test_assemble_path_at_exsting_folder(mocker):
     target_folder = 'admin/test_folder_exist'
     project_code = 'test_project'
     zone = 0
-    resumable_id = None
 
     node_list = [
         {
@@ -74,7 +73,7 @@ def test_assemble_path_at_exsting_folder(mocker):
     mocker.patch('app.services.file_manager.file_upload.file_upload.search_item', side_effect=node_list)
 
     current_file_path, parent_folder, create_folder_flag, _ = assemble_path(
-        local_file_path, target_folder, project_code, zone, resumable_id
+        local_file_path, target_folder, project_code, FolderType.NAMEFOLDER, zone
     )
     assert current_file_path == 'admin/test_folder_exist/file.txt'
     assert parent_folder.get('name') == 'test_folder_exist'
@@ -86,7 +85,6 @@ def test_assemble_path_at_non_existing_folder(mocker):
     target_folder = 'admin/test_folder_not_exist'
     project_code = 'test_project'
     zone = 0
-    resumable_id = None
 
     node_list = [
         {
@@ -106,7 +104,7 @@ def test_assemble_path_at_non_existing_folder(mocker):
     mocker.patch('app.services.file_manager.file_upload.file_upload.click.confirm', return_value=None)
 
     current_file_path, parent_folder, create_folder_flag, _ = assemble_path(
-        local_file_path, target_folder, project_code, zone, resumable_id
+        local_file_path, target_folder, project_code, FolderType.NAMEFOLDER, zone
     )
     assert current_file_path == 'admin/test_folder_not_exist'
     assert parent_folder.get('name') == 'admin'
@@ -118,7 +116,6 @@ def test_assemble_path_at_project_folder(mocker):
     target_folder = 'project_folder'
     project_code = 'test_project'
     zone = 0
-    resumable_id = None
 
     mocker.patch(
         'app.services.file_manager.file_upload.file_upload.search_item',
@@ -135,7 +132,7 @@ def test_assemble_path_at_project_folder(mocker):
     )
 
     current_file_path, parent_folder, create_folder_flag, target_folder = assemble_path(
-        local_file_path, target_folder, project_code, zone, resumable_id
+        local_file_path, target_folder, project_code, FolderType.PROJECTFOLDER, zone
     )
     assert current_file_path == 'shared/project_folder/file.txt'
     assert parent_folder.get('name') == 'project_folder'

tests/app/utils/test_aggregated.py

@@ -5,7 +5,9 @@
 import pytest
 
 from app.configs.app_config import AppConfig
+from app.models.folder import FolderType
 from app.utils.aggregated import check_item_duplication
+from app.utils.aggregated import identify_target_folder
 from app.utils.aggregated import search_item
 from app.utils.aggregated import validate_folder_name
 from tests.conftest import decoded_token
@@ -129,3 +131,24 @@ def test_check_duplicate_fail_with_error_code(httpx_mock, mocker, capsys):
 def test_validate_folder_name(folder_name):
     valid = validate_folder_name(folder_name)
     assert valid is False
+
+
+@pytest.mark.parametrize(
+    'input_path,expected_result',
+    [
+        ('project_code/username', ('project_code', FolderType.NAMEFOLDER, 'username')),
+        ('project_code/username/folder1', ('project_code', FolderType.NAMEFOLDER, 'username/folder1')),
+        ('project_code/projectfolder/folder1', ('project_code', FolderType.PROJECTFOLDER, 'folder1')),
+        ('project_code/projectfolder/folder1/folder2', ('project_code', FolderType.PROJECTFOLDER, 'folder1/folder2')),
+    ],
+)
+def test_identify_target_folder_success_with_different_path(mocker, input_path, expected_result):
+    mocker.patch('app.utils.aggregated.validate_folder_name', return_value=True)
+    result = identify_target_folder(input_path)
+    assert result == expected_result
+
+
+def test_identify_target_folder_fail_with_invalid_input(mocker):
+    mocker.patch('app.utils.aggregated.validate_folder_name', return_value=False)
+    with pytest.raises(SystemExit):
+        identify_target_folder('project_code')