Add allow and ignore list of processes in dispatcher

docs/CHANGELOG.rst

@@ -14,6 +14,11 @@ Changed
 -------
 - Redis cache in listener is updated when data fields are retrieved from the
   database
+Added
+-----
+- Add processes allow and ignore list to dispatcher, controlled by
+  environmental variables ``FLOW_PROCESSES_ALLOW_LIST`` and
+  ``FLOW_PROCESSES_IGNORE_LIST```
 
 Fixed
 -----

resolwe/flow/managers/dispatcher.py

@@ -166,6 +166,10 @@ def __init__(self, *args, **kwargs):
         else:
             connector_list = [flow_manager.get("NAME", DEFAULT_CONNECTOR)]
 
+        # Store the whitelist and blacklist for later use.
+        self._processes_allow = settings.FLOW_PROCESSES_ALLOW_LIST
+        self._processes_ignore = settings.FLOW_PROCESSES_IGNORE_LIST
+
         # Pre-load all needed connectors.
         self.connectors = {}
         for module_name in connector_list:
@@ -669,6 +673,13 @@ def process_data_object(data: Data):
 
         try:
             queryset = Data.objects.filter(status=Data.STATUS_RESOLVING)
+            # Check if process is in the whitelist or blacklist. The blacklist has
+            # priority.
+            if self._processes_allow:
+                queryset = queryset.filter(process__slug__in=self._processes_allow)
+            if self._processes_ignore:
+                queryset = queryset.exclude(process__slug__in=self._processes_ignore)
+
             if data_id is not None:
                 # Scan only given data object and its children.
                 queryset = queryset.filter(

resolwe/flow/tests/test_manager.py

@@ -23,8 +23,9 @@ def setUp(self):
         super().setUp()
 
         self.collection = Collection.objects.create(contributor=self.contributor)
-
         self._register_schemas(processes_paths=[PROCESSES_DIR])
+        manager._processes_ignore = None
+        manager._processes_allow = None
 
     def test_create_data(self):
         """Test that manager is run when new object is created."""
@@ -34,10 +35,44 @@ def test_create_data(self):
             contributor=self.contributor,
             process=process,
         )
-
         data.refresh_from_db()
         self.assertEqual(data.status, Data.STATUS_DONE)
 
+    def test_ignore_allow_list(self):
+        process = Process.objects.filter(slug="test-min").latest()
+
+        # Ignored processes should not trigger processing.
+        manager._processes_ignore = ["test-min"]
+        data = Data.objects.create(
+            name="Test data",
+            contributor=self.contributor,
+            process=process,
+        )
+        data.refresh_from_db()
+        self.assertEqual(data.status, Data.STATUS_RESOLVING)
+
+        # Ignore should have precedence.
+        manager._processes_ignore = ["test-min"]
+        manager._processes_allow = ["test-min"]
+        data = Data.objects.create(
+            name="Test data",
+            contributor=self.contributor,
+            process=process,
+        )
+        data.refresh_from_db()
+        self.assertEqual(data.status, Data.STATUS_RESOLVING)
+
+        # Allowing some processes shoud disable others.
+        manager._processes_ignore = None
+        manager._processes_allow = ["test-something-else"]
+        data = Data.objects.create(
+            name="Test data",
+            contributor=self.contributor,
+            process=process,
+        )
+        data.refresh_from_db()
+        self.assertEqual(data.status, Data.STATUS_RESOLVING)
+
     def test_spawned_process(self):
         """Test that manager is run for spawned processes and permissions are copied."""
         DescriptorSchema.objects.create(

setup.py

@@ -57,6 +57,7 @@
         "jsonschema~=4.17.3",
         "plumbum~=1.8.1",
         "psycopg[binary]~=3.1.9",
+        "python-decouple~=3.8",
         "PyYAML~=6.0",
         "redis~=4.5.4",
         "shellescape~=3.8.1",

tests/settings.py

@@ -2,11 +2,14 @@
 Django settings for running tests for Resolwe package.
 
 """
+import json
 import os
 import sys
 from distutils.util import strtobool
 from pathlib import Path
 
+from decouple import Csv, config
+
 from resolwe.__about__ import __version__
 
 PROJECT_ROOT = Path(__file__).parent.resolve()
@@ -204,6 +207,16 @@
 
 FLOW_PROCESSES_RUNTIMES = ("resolwe.process.runtime.Process",)
 
+# Which processes to schedule (if not set, all).
+FLOW_PROCESSES_ALLOW_LIST = config(
+    "FLOW_PROCESSES_ALLOW_LIST", default="null", cast=json.loads
+)
+
+# Which processes to ignore (by default, none).
+FLOW_PROCESSES_IGNORE_LIST = config(
+    "FLOW_PROCESSES_IGNORE_LIST", default="null", cast=json.loads
+)
+
 FLOW_EXECUTOR = {
     "NAME": "resolwe.flow.executors.docker",
     "LISTENER_CONNECTION": LISTENER_CONNECTION,