add BatchCheckFileSuffixTask and tests (#2064)

Author: mikkonie

CHANGELOG.rst

@@ -16,12 +16,14 @@ Added
     - ``PROJECTROLES_SUPPORT_CONTACT`` setting support (#2095)
 - **Landingzones**
     - Site read-only mode support (#2051)
+    - ``file_name_prohibit`` app setting (#2064)
 - **Samplesheets**
     - Site read-only mode support (#2051)
 - **Taskflowbackend**
     - Project deletion support (#2051)
     - Zone validation and moving progress indicators (#2024)
     - ``TASKFLOW_ZONE_PROGRESS_INTERVAL`` Django setting (#2024)
+    - ``BatchCheckFileSuffixTask`` iRODS task (#2064)
 
 Changed
 -------
@@ -36,6 +38,7 @@ Changed
     - Define app settings as ``PluginAppSettingDef`` objects (#2051)
 - **Taskflowbackend**
     - Enable no role for old owner in ``perform_owner_transfer()`` (#2051)
+    - Rename ``BatchCheckFileTask`` to ``BatchCheckFileExistTask`` (#2064)
 
 Fixed
 -----

docs_manual/source/sodar_release_notes.rst

@@ -14,6 +14,7 @@ v1.1.0 (WIP)
 Release for SODAR Core 1.1 upgrade and feature updates.
 
 - Add landing zone validation and moving progress indicators
+- Add landing zone file type prohibiting
 - Upgrade to python-irodsclient v3.0
 - Upgrade to SODAR Core v1.1.2
 - Replace SODAR_SUPPORT_* settings with PROJECTROLES_SUPPORT_CONTACT

landingzones/plugins.py

@@ -62,7 +62,7 @@
         user_modifiable=True,
     ),
     PluginAppSettingDef(
-        name='file_type_prohibit',
+        name='file_name_prohibit',
         scope=APP_SETTING_SCOPE_PROJECT,
         type=APP_SETTING_TYPE_STRING,
         default='',

taskflowbackend/tasks/irods_tasks.py

@@ -39,6 +39,7 @@
 INHERIT_STRINGS = {True: 'inherit', False: 'noinherit'}
 META_EMPTY_VALUE = 'N/A'
 MD5_RE = re.compile(r'([^\w.])')
+SUFFIX_CLEAN_RE = re.compile(r'\A\W+|\W+\Z')
 CHECKSUM_RETRY = 5
 NO_FILE_CHECKSUM_LABEL = 'None'
 
@@ -614,7 +615,36 @@ def revert(
             self.revert_set_access(path, user_name, obj_target, recursive)
 
 
-class BatchCheckFilesTask(IrodsBaseTask):
+class BatchCheckFileSuffixTask(IrodsBaseTask):
+    """Batch check for prohibited file name suffixes"""
+
+    def execute(self, file_paths, suffixes, zone_path, *args, **kwargs):
+        # Cleanup suffixes
+        suffixes = [
+            re.sub(SUFFIX_CLEAN_RE, '', s.lower().strip())
+            for s in suffixes
+            if not s.lower().endswith('md5')
+        ]
+        err_paths = []
+        for p in file_paths:
+            if any(p.lower().endswith('.' + s) for s in suffixes):
+                err_paths.append(p)
+        err_len = len(err_paths)
+        if err_len > 0:
+            msg = '{} file{} of prohibited type found: {}'.format(
+                err_len,
+                's' if err_len != 1 else '',
+                ';'.join([p.replace(zone_path + '/', '') for p in err_paths]),
+            )
+            logger.error(msg)
+            self._raise_irods_exception(Exception(), msg)
+        super().execute(*args, **kwargs)
+
+    def revert(self, file_paths, suffixes, zone_path, *args, **kwargs):
+        pass  # Nothing to revert
+
+
+class BatchCheckFileExistTask(IrodsBaseTask):
     """
     Batch check for existence of files and corresponding .md5 checksum files
     """

taskflowbackend/tests/test_tasks.py

@@ -21,6 +21,7 @@
     ZONE_TITLE,
     ZONE_DESC,
 )
+from landingzones.tests.test_views_taskflow import LandingZoneTaskflowMixin
 
 # Samplesheets dependency
 from samplesheets.tests.test_io import SampleSheetIOMixin, SHEET_DIR
@@ -67,6 +68,10 @@
 BATCH_OBJ_NAME = 'batch_obj'
 BATCH_OBJ2_NAME = 'batch_obj2'
 
+SUFFIX_OBJ_NAME_BAM = 'test.bam'
+SUFFIX_OBJ_NAME_VCF = 'test.vcf.gz'
+SUFFIX_OBJ_NAME_TXT = ' test.txt'
+
 
 class IRODSTaskTestBase(TaskflowViewTestBase):
     """Base test class for iRODS tasks"""
@@ -1868,6 +1873,180 @@ def test_revert_mixed(self):
         self.assert_irods_access(DEFAULT_USER_GROUP, self.sub_coll_path2, None)
 
 
+class TestBatchCheckFileSuffixTask(
+    SampleSheetIOMixin,
+    LandingZoneMixin,
+    LandingZoneTaskflowMixin,
+    IRODSTaskTestBase,
+):
+    """Tests for BatchCheckFileSuffixTask"""
+
+    def setUp(self):
+        super().setUp()
+        self.investigation = self.import_isa_from_file(SHEET_PATH, self.project)
+        self.study = self.investigation.studies.first()
+        self.assay = self.study.assays.first()
+        self.zone = self.make_landing_zone(
+            title=ZONE_TITLE,
+            project=self.project,
+            user=self.user,
+            assay=self.assay,
+            description=ZONE_DESC,
+        )
+        self.make_zone_taskflow(zone=self.zone)
+        self.zone_path = self.irods_backend.get_path(self.zone)
+        self.zone_coll = self.irods.collections.get(self.zone_path)
+        self.obj_bam = self.make_irods_object(
+            self.zone_coll, SUFFIX_OBJ_NAME_BAM
+        )
+        self.obj_vcf = self.make_irods_object(
+            self.zone_coll, SUFFIX_OBJ_NAME_VCF
+        )
+        self.obj_txt = self.make_irods_object(
+            self.zone_coll, SUFFIX_OBJ_NAME_TXT
+        )
+        self.obj_paths = [
+            self.obj_bam.path,
+            self.obj_vcf.path,
+            self.obj_txt.path,
+        ]
+        self.task_kw = {
+            'cls': BatchCheckFileSuffixTask,
+            'name': 'Check file suffixes',
+            'inject': {
+                'file_paths': self.obj_paths,
+                'zone_path': self.zone_path,
+            },
+        }
+
+    def test_check_bam(self):
+        """Test batch file suffix check with prohibited BAM type"""
+        self.task_kw['inject']['suffixes'] = ['bam']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_vcf(self):
+        """Test check with prohibited VCF type"""
+        self.task_kw['inject']['suffixes'] = ['vcf.gz']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertNotIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_multiple(self):
+        """Test check with multiple prohibited types"""
+        self.task_kw['inject']['suffixes'] = ['bam', 'vcf.gz']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_multiple_not_found(self):
+        """Test check with multiple types not found in files"""
+        self.task_kw['inject']['suffixes'] = ['mp3', 'rar']
+        self.add_task(**self.task_kw)
+        result = self.run_flow()
+        self.assertEqual(result, True)
+
+    def test_check_empty_list(self):
+        """Test check with empty prohibition list"""
+        self.task_kw['inject']['suffixes'] = []
+        self.add_task(**self.task_kw)
+        result = self.run_flow()
+        self.assertEqual(result, True)
+
+    def test_check_notation_dot(self):
+        """Test check with dot notation in list"""
+        self.task_kw['inject']['suffixes'] = ['.bam']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_notation_asterisk(self):
+        """Test check with asterisk notation in list"""
+        self.task_kw['inject']['suffixes'] = ['*bam']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_notation_combined(self):
+        """Test check with combined notation in list"""
+        self.task_kw['inject']['suffixes'] = ['*.bam']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_extra_spaces(self):
+        """Test check with extra spaces"""
+        self.task_kw['inject']['suffixes'] = [' bam ']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_not_end_of_file(self):
+        """Test check with given string not in end of file name"""
+        self.task_kw['inject']['suffixes'] = ['test']
+        self.add_task(**self.task_kw)
+        result = self.run_flow()
+        self.assertEqual(result, True)
+
+    def test_check_upper_case(self):
+        """Test check with upper case string"""
+        self.task_kw['inject']['suffixes'] = ['BAM']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+    def test_check_invalid_strings(self):
+        """Test check with invalid strings"""
+        self.task_kw['inject']['suffixes'] = ['', '*', '*.*']
+        self.add_task(**self.task_kw)
+        result = self.run_flow()
+        self.assertEqual(result, True)
+
+    def test_check_invalid_valid(self):
+        """Test check with mixed invalid and valid strings"""
+        self.task_kw['inject']['suffixes'] = ['', '*', 'bam']
+        self.add_task(**self.task_kw)
+        with self.assertRaises(Exception) as cm:
+            self.run_flow()
+            ex = cm.exception
+            self.assertIn(SUFFIX_OBJ_NAME_BAM, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_VCF, ex)
+            self.assertNotIn(SUFFIX_OBJ_NAME_TXT, ex)
+
+
 class TestBatchCreateCollectionsTask(IRODSTaskTestBase):
     """Tests for BatchCreateCollectionsTask"""