add sheet sync setting validation (wip) (#1310, #1384)

CHANGELOG.rst

@@ -13,6 +13,8 @@ Added
 
 - **Irodsbackend**
     - ``get_version()`` helper (#1592, #1817)
+- **Samplesheets**
+    - ``sheet_sync_url`` and ``sheet_sync_token`` custom validation (#1310, #1384)
 
 Changed
 -------

docs_manual/source/sodar_release_notes.rst

@@ -14,6 +14,7 @@ v0.14.1 (WIP)
 Release for minor updates, maintenance and bug fixes.
 
 - Add iRODS v4.3 support
+- Add sheet sync setting validation
 - Change default IGV genome to "b37_1kg"
 - Fix landing zone locking controls for non-superusers
 - Minor updates and bug fixes

requirements/base.txt

@@ -71,7 +71,7 @@ django-autocomplete-light==3.9.4
 
 # SODAR Core
 # django-sodar-core==0.13.2
--e git+https://github.com/bihealth/sodar-core.git@4657a3c9e24fa338444e44a0865996fd63abadc3#egg=django-sodar-core
+-e git+https://github.com/bihealth/sodar-core.git@10f85429c1a84ce40a28b096b560b4c51430be3b#egg=django-sodar-core
 
 # Celery
 celery==5.2.7

samplesheets/plugins.py

@@ -2,10 +2,12 @@
 
 import logging
 import os
+import re
 import time
 
 from copy import deepcopy
 from irods.exception import NetworkException
+from urllib.parse import urlparse
 
 from django.conf import settings
 from django.urls import reverse
@@ -79,6 +81,10 @@
 SKIP_MSG_NO_INV = 'No investigation for project'
 SKIP_MSG_NO_COLLS = 'Investigation collections not created in iRODS'
 IGV_DEFAULT_GENOME = 'b37_1kg'
+SYNC_URL_RE = (
+    r'/samplesheets/sync/[a-f0-9]{8}-?[a-f0-9]{4}-?4[a-f0-9]{3}-?[89ab][a-f0-9]'
+    r'{3}-?[a-f0-9]{12}'
+)
 
 
 # Samplesheets project app plugin ----------------------------------------------
@@ -529,6 +535,38 @@ def get_project_list_value(self, column_id, project, user):
                 )
             )
 
+    # TODO: Add tests
+    def validate_form_app_settings(self, app_settings, project=None, user=None):
+        """
+        Validate app settings form data and return a dict of errors.
+
+        :param app_settings: Dict of app settings
+        :param project: Project object or None
+        :param user: User object
+        :return: dict in format of {setting_name: 'Error string'}
+        """
+        ret = {}
+        # Ensure all sheet sync settings are set if sync is enabled
+        if project and app_settings.get('sheet_sync_enable') is not None:
+            enable = app_settings['sheet_sync_enable']
+            msg = 'must be set if sheet sync is enabled'
+            if enable and not app_settings['sheet_sync_token']:
+                ret['sheet_sync_token'] = 'Token ' + msg
+            if enable and not app_settings['sheet_sync_url']:
+                ret['sheet_sync_url'] = 'URL ' + msg
+        if project and app_settings.get('sheet_sync_url'):
+            try:
+                urlparse(app_settings['sheet_sync_url'])
+            except Exception:
+                ret['sheet_sync_url'] = 'Invalid URL'
+            if 'sheet_sync_url' not in ret and not re.findall(
+                SYNC_URL_RE, app_settings['sheet_sync_url']
+            ):
+                ret[
+                    'sheet_sync_url'
+                ] = 'URL does not point to a sheet sync endpoint'
+        return ret
+
     # Project Modify API Implementation ----------------------------------------
 
     @classmethod

samplesheets/tests/test_views_taskflow.py

@@ -1,6 +1,7 @@
 """View tests in the samplesheets Django app with taskflow"""
 
 import os
+import uuid
 
 from datetime import timedelta
 from irods.exception import CollectionDoesNotExist, NoResultFound
@@ -71,10 +72,10 @@
 
 # Local constants
 APP_NAME = 'samplesheets'
+DUMMY_UUID = str(uuid.uuid4())
 SHEET_PATH = SHEET_DIR + 'i_small.zip'
 IRODS_FILE_NAME = 'test1.txt'
 IRODS_FILE_NAME2 = 'test2.txt'
-DUMMY_UUID = 'aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa'
 PUBLIC_USER_NAME = 'user_no_roles'
 PUBLIC_USER_PASS = 'password'
 SOURCE_ID = '0815'
@@ -84,6 +85,9 @@
 TICKET_LABEL_UPDATED = 'TestTicketUpdated'
 TICKET_STR = 'q657xxx3i2x2b8vj'
 IRODS_REQUEST_DESC_UPDATE = 'Updated'
+SHEET_SYNC_URL = 'https://sodar.instance/samplesheets/sync/' + DUMMY_UUID
+SHEET_SYNC_URL_INVALID = 'https://some.sodar/not-valid-url'
+SHEET_SYNC_TOKEN = 'dohdai4EZie0xooF'
 
 
 class SampleSheetTaskflowMixin:
@@ -2573,3 +2577,152 @@ def test_search_limit_file(self):
         self.assertEqual(
             data['results']['files']['items'][0]['name'], self.file_name
         )
+
+
+class TestProjectUpdateView(TaskflowViewTestBase):
+    """Tests for samplesheets app settings in ProjectUpdateView"""
+
+    def setUp(self):
+        super().setUp()
+        self.project, self.owner_as = self.make_project_taskflow(
+            title='TestProject',
+            type=PROJECT_TYPE_PROJECT,
+            parent=self.category,
+            owner=self.user,
+            description='description',
+        )
+        self.values = model_to_dict(self.project)
+        self.values['parent'] = self.category.sodar_uuid
+        self.values['owner'] = self.user.sodar_uuid
+        self.values.update(
+            app_settings.get_all(project=self.project, post_safe=True)
+        )
+        self.url = reverse(
+            'projectroles:update',
+            kwargs={'project': self.project.sodar_uuid},
+        )
+
+    def test_post_sync_default(self):
+        """Test POST with default sheet sync values"""
+        self.values['description'] = 'updated description'
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 302)
+        self.assertFalse(
+            app_settings.get(
+                APP_NAME, 'sheet_sync_enable', project=self.project
+            )
+        )
+        self.assertEqual(
+            app_settings.get(APP_NAME, 'sheet_sync_url', project=self.project),
+            '',
+        )
+        self.assertEqual(
+            app_settings.get(
+                APP_NAME, 'sheet_sync_token', project=self.project
+            ),
+            '',
+        )
+
+    def test_post_sync_enable(self):
+        """Test POST with enabled sync and correct url/token"""
+        self.values['description'] = 'updated description'
+        self.assertFalse(
+            app_settings.get(
+                APP_NAME, 'sheet_sync_enable', project=self.project
+            )
+        )
+        self.assertEqual(
+            app_settings.get(APP_NAME, 'sheet_sync_url', project=self.project),
+            '',
+        )
+        self.assertEqual(
+            app_settings.get(
+                APP_NAME, 'sheet_sync_token', project=self.project
+            ),
+            '',
+        )
+
+        self.values['settings.samplesheets.sheet_sync_enable'] = True
+        self.values['settings.samplesheets.sheet_sync_url'] = SHEET_SYNC_URL
+        self.values['settings.samplesheets.sheet_sync_token'] = SHEET_SYNC_TOKEN
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+            self.assertRedirects(
+                response,
+                reverse(
+                    'projectroles:detail',
+                    kwargs={'project': self.project.sodar_uuid},
+                ),
+            )
+
+        self.assertEqual(response.status_code, 302)
+        self.project.refresh_from_db()
+        self.assertEqual(self.project.description, 'updated description')
+        self.assertTrue(
+            app_settings.get(
+                APP_NAME, 'sheet_sync_enable', project=self.project
+            )
+        )
+        self.assertEqual(
+            app_settings.get(APP_NAME, 'sheet_sync_url', project=self.project),
+            SHEET_SYNC_URL,
+        )
+        self.assertEqual(
+            app_settings.get(
+                APP_NAME, 'sheet_sync_token', project=self.project
+            ),
+            SHEET_SYNC_TOKEN,
+        )
+
+    def test_post_sync_no_url_or_token(self):
+        """Test POST with enabled sync and no URL or token"""
+        self.values['settings.samplesheets.sheet_sync_enable'] = True
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 200)
+
+    def test_post_sync_no_token(self):
+        """Test POST with enabled sync and no token"""
+        self.values['settings.samplesheets.sheet_sync_enable'] = True
+        self.values['settings.samplesheets.sheet_sync_url'] = SHEET_SYNC_URL
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 200)
+
+    def test_post_sync_no_url(self):
+        """Test POST with enabled sync and no URL or token"""
+        self.values['settings.samplesheets.sheet_sync_enable'] = True
+        self.values['settings.samplesheets.sheet_sync_token'] = SHEET_SYNC_TOKEN
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 200)
+
+    def test_post_sync_invalid_url(self):
+        """Test POST with enabled sync and no token"""
+        self.values['settings.samplesheets.sheet_sync_enable'] = True
+        self.values[
+            'settings.samplesheets.sheet_sync_url'
+        ] = SHEET_SYNC_URL_INVALID
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 200)
+
+    def test_post_sync_disabled(self):
+        """Test POST with disabled sync and valid input"""
+        self.values['settings.samplesheets.sheet_sync_enable'] = False
+        self.values['settings.samplesheets.sheet_sync_url'] = SHEET_SYNC_URL
+        self.values['settings.samplesheets.sheet_sync_token'] = SHEET_SYNC_TOKEN
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 302)
+
+    def test_post_sync_disabled_invalid_url(self):
+        """Test POST with disabled sync and invalid input"""
+        self.values['settings.samplesheets.sheet_sync_enable'] = False
+        self.values[
+            'settings.samplesheets.sheet_sync_url'
+        ] = SHEET_SYNC_URL_INVALID
+        with self.login(self.user):
+            response = self.client.post(self.url, self.values)
+        self.assertEqual(response.status_code, 200)