Enabled asynchronous learning from user using RQ

.github/workflows/plugin.yml

@@ -36,6 +36,15 @@ jobs:
         ports:
           - 3306/tcp
         options: --health-cmd="mysqladmin ping" --health-interval=10s --health-timeout=5s --health-retries=3
+      redis:
+        image: redis
+        ports:
+          - 6379/tcp
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
 
     strategy:
       matrix:
@@ -51,7 +60,7 @@ jobs:
           python-version: ${{ matrix.python-version }}
       - name: Install dependencies
         run: |
-          sudo apt-get update -y && sudo apt-get install -y librrd-dev rrdtool
+          sudo apt-get update -y && sudo apt-get install -y librrd-dev rrdtool redis-server
           python -m pip install --upgrade pip
           #pip install -e git+https://github.com/modoboa/modoboa.git#egg=modoboa
           pip install -r requirements.txt
@@ -62,6 +71,11 @@ jobs:
           python setup.py develop
           cd ../modoboa-amavis
           python setup.py develop
+          echo "Testing redis connection"
+          redis-cli -h $REDIS_HOST -p $REDIS_PORT ping
+        env:
+          REDIS_HOST: localhost
+          REDIS_PORT: ${{ job.services.redis.ports[6379] }}
       - name: Install postgres requirements
         if: ${{ matrix.database == 'postgres' }}
         run: |
@@ -85,7 +99,8 @@ jobs:
           MYSQL_HOST: 127.0.0.1
           MYSQL_PORT: ${{ job.services.mysql.ports[3306] }} # get randomly assigned published port
           MYSQL_USER: root
-
+          REDIS_HOST: localhost
+          REDIS_PORT: ${{ job.services.redis.ports[6379] }}
       - name: Test with pytest and coverage
         if: ${{ matrix.python-version == '3.10' && matrix.database == 'postgres' }}
         run: |
@@ -100,6 +115,8 @@ jobs:
           MYSQL_HOST: 127.0.0.1
           MYSQL_PORT: ${{ job.services.mysql.ports[3306] }} # get randomly assigned published port
           MYSQL_USER: root
+          REDIS_HOST: localhost
+          REDIS_PORT: ${{ job.services.redis.ports[6379] }}
       - name: Upload coverage result
         if: ${{ matrix.python-version == '3.10' && matrix.database == 'postgres' }}
         uses: actions/upload-artifact@v3

modoboa_amavis/tasks.py

@@ -0,0 +1,37 @@
+"""Async tasks."""
+
+from typing import List
+
+from django.utils.translation import ngettext
+
+from modoboa.core import models as core_models
+
+from .lib import SpamassassinClient
+from .sql_connector import SQLconnector
+
+
+def manual_learning(user_pk: int,
+                    mtype: str,
+                    selection: List[str],
+                    recipient_db: str):
+    """Task to trigger manual learning for given selection."""
+    user = core_models.User.objects.get(pk=user_pk)
+    connector = SQLconnector()
+    saclient = SpamassassinClient(user, recipient_db)
+    for item in selection:
+        rcpt, mail_id = item.split()
+        content = connector.get_mail_content(mail_id.encode("ascii"))
+        result = saclient.learn_spam(rcpt, content) if mtype == "spam" \
+            else saclient.learn_ham(rcpt, content)
+        if not result:
+            break
+        connector.set_msgrcpt_status(
+            rcpt, mail_id, mtype[0].upper()
+        )
+    if saclient.error is None:
+        saclient.done()
+        message = ngettext("%(count)d message processed successfully",
+                           "%(count)d messages processed successfully",
+                           len(selection)) % {"count": len(selection)}
+    else:
+        message = saclient.error

modoboa_amavis/tests/test_views.py

@@ -5,11 +5,15 @@
 import os
 from unittest import mock
 
+from rq import SimpleWorker
+
 from django.core import mail
 from django.core.management import call_command
 from django.test import override_settings
 from django.urls import reverse
 
+import django_rq
+
 from modoboa.admin import factories as admin_factories
 from modoboa.core import models as core_models
 from modoboa.lib.tests import ModoTestCase
@@ -226,7 +230,10 @@ def _test_mark_message(self, action, status):
         data = {"rcpt": smart_str(self.msgrcpt.rid.email)}
         response = self.ajax_post(url, data)
         self.assertEqual(
-            response["message"], "1 message processed successfully")
+            response["message"], "Your request is being processed...")
+        queue = django_rq.get_queue("default")
+        worker = SimpleWorker([queue], connection=queue.connection)
+        worker.work(burst=True)
         self.msgrcpt.refresh_from_db()
         self.assertEqual(self.msgrcpt.rs, status)
 
@@ -235,7 +242,8 @@ def _test_mark_message(self, action, status):
         self.set_global_parameter("sa_is_local", False)
         response = self.ajax_post(url, data)
         self.assertEqual(
-            response["message"], "1 message processed successfully")
+            response["message"], "Your request is being processed...")
+        worker.work(burst=True)
         self.msgrcpt.refresh_from_db()
         self.assertEqual(self.msgrcpt.rs, status)
 
@@ -289,7 +297,7 @@ def test_manual_learning_as_user(self):
         self._test_mark_message("ham", "H")
 
     @mock.patch("socket.socket")
-    def test_process(self, mock_socket):
+    def test_process_release(self, mock_socket):
         """Test process mode (bulk)."""
         # Initiate session
         url = reverse("modoboa_amavis:_mail_list")
@@ -305,8 +313,9 @@ def test_process(self, mock_socket):
                 smart_str(msgrcpt.rid.email),
                 smart_str(msgrcpt.mail.mail_id)),
         ]
-        mock_socket.return_value.recv.side_effect = (
-            b"250 1234 Ok\r\n", b"250 1234 Ok\r\n")
+        mock_socket.return_value.recv.side_effect = [
+            b"250 1234 Ok\r\n", b"250 1234 Ok\r\n"
+        ]
         data = {
             "action": "release",
             "rcpt": smart_str(self.msgrcpt.rid.email),
@@ -317,15 +326,36 @@ def test_process(self, mock_socket):
         self.assertEqual(
             response["message"], "2 messages released successfully")
 
+    def test_process_all(self):
+        """Test process mode (bulk)."""
+        # Initiate session
+        url = reverse("modoboa_amavis:_mail_list")
+        response = self.ajax_get(url)
+
+        msgrcpt = factories.create_spam("[email protected]")
+        url = reverse("modoboa_amavis:mail_process")
+        selection = [
+            "{} {}".format(
+                smart_str(self.msgrcpt.rid.email),
+                smart_str(self.msgrcpt.mail.mail_id)),
+            "{} {}".format(
+                smart_str(msgrcpt.rid.email),
+                smart_str(msgrcpt.mail.mail_id)),
+        ]
+        data = {
+            "rcpt": smart_str(self.msgrcpt.rid.email),
+            "selection": ",".join(selection)
+        }
+
         data["action"] = "mark_as_spam"
         response = self.ajax_post(url, data)
         self.assertEqual(
-            response["message"], "2 messages processed successfully")
+            response["message"], "Your request is being processed...")
 
         data["action"] = "mark_as_ham"
         response = self.ajax_post(url, data)
         self.assertEqual(
-            response["message"], "2 messages processed successfully")
+            response["message"], "Your request is being processed...")
 
         data = {
             "action": "delete",

modoboa_amavis/views.py

@@ -14,15 +14,18 @@
 from django.utils.translation import gettext as _, ngettext
 from django.views.decorators.csrf import csrf_exempt
 
+import django_rq
+
 from modoboa.admin.models import Domain, Mailbox
+from modoboa_amavis import tasks
 from modoboa.lib.exceptions import BadRequest
 from modoboa.lib.paginator import Paginator
 from modoboa.lib.web_utils import getctx, render_to_json_response
 from modoboa.parameters import tools as param_tools
 from . import constants
 from .forms import LearningRecipientForm
 from .lib import (
-    AMrelease, QuarantineNavigationParameters, SpamassassinClient,
+    AMrelease, QuarantineNavigationParameters,
     manual_learning_enabled, selfservice
 )
 from .models import Msgrcpt, Msgs
@@ -192,7 +195,7 @@ def viewheaders(request, mail_id):
     return render(request, "modoboa_amavis/viewheader.html", context)
 
 
-def check_mail_id(request, mail_id):
+def check_mail_id(request, mail_id) -> list:
     if isinstance(mail_id, six.string_types):
         if "rcpt" in request.POST:
             mail_id = ["%s %s" % (request.POST["rcpt"], mail_id)]
@@ -345,29 +348,17 @@ def mark_messages(request, selection, mtype, recipient_db=None):
             "user" if request.user.role == "SimpleUsers" else "global"
         )
     selection = check_mail_id(request, selection)
-    connector = SQLconnector()
-    saclient = SpamassassinClient(request.user, recipient_db)
-    for item in selection:
-        rcpt, mail_id = item.split()
-        content = connector.get_mail_content(mail_id.encode("ascii"))
-        result = saclient.learn_spam(rcpt, content) if mtype == "spam" \
-            else saclient.learn_ham(rcpt, content)
-        if not result:
-            break
-        connector.set_msgrcpt_status(
-            rcpt, mail_id, mtype[0].upper()
-        )
-    if saclient.error is None:
-        saclient.done()
-        message = ngettext("%(count)d message processed successfully",
-                            "%(count)d messages processed successfully",
-                            len(selection)) % {"count": len(selection)}
-    else:
-        message = saclient.error
-    status = 400 if saclient.error else 200
+
+    queue = django_rq.get_queue("default")
+    queue.enqueue(tasks.manual_learning,
+                  request.user.pk,
+                  mtype,
+                  selection,
+                  recipient_db)
+
     return render_to_json_response({
-        "message": message, "reload": True
-    }, status=status)
+        "message": _("Your request is being processed..."), "reload": True
+    })
 
 
 @login_required

test_project/test_project/settings.py

@@ -189,6 +189,21 @@
 
 MODOBOA_API_URL = "https://api.modoboa.org/1/"
 
+# REDIS
+
+REDIS_HOST = os.environ.get('REDIS_HOST', '127.0.0.1')
+REDIS_PORT = os.environ.get('REDIS_PORT', 6379)
+REDIS_QUOTA_DB = 0
+REDIS_URL = 'redis://{}:{}/{}'.format(REDIS_HOST, REDIS_PORT, REDIS_QUOTA_DB)
+
+# RQ
+
+RQ_QUEUES = {
+    'default': {
+        'URL': REDIS_URL,
+    },
+}
+
 # Password validation
 # https://docs.djangoproject.com/en/1.11/ref/settings/#auth-password-validators