refactor: sync to RFC Editor queue via celery

ietf/sync/tasks.py

@@ -13,6 +13,7 @@
 
 from ietf.sync import iana
 from ietf.sync import rfceditor
+from ietf.sync.rfceditor import MIN_QUEUE_RESULTS, parse_queue, update_drafts_from_queue
 from ietf.utils import log
 from ietf.utils.timezone import date_today
 
@@ -70,6 +71,33 @@ def rfc_editor_index_update_task(full_index=False):
             log.log("RFC%s, %s: %s" % (rfc_number, doc.name, c))
 
 
+@shared_task
+def rfc_editor_queue_updates_task():
+    log.log(f"Updating RFC Editor queue states from {settings.RFC_EDITOR_QUEUE_URL}")
+    try:
+        response = requests.get(
+            settings.RFC_EDITOR_QUEUE_URL,
+            timeout=30,  # seconds
+        )
+    except requests.Timeout as exc:
+        log.log(f"GET request timed out retrieving RFC editor queue: {exc}")
+        return  # failed
+    drafts, warnings = parse_queue(io.StringIO(response.text))
+    for w in warnings:
+        log.log(f"Warning: {w}")
+
+    if len(drafts) < MIN_QUEUE_RESULTS:
+        log.log("Not enough results, only %s" % len(drafts))
+        return  # failed
+
+    changed, warnings = update_drafts_from_queue(drafts)
+    for w in warnings:
+        log.log(f"Warning: {w}")
+
+    for c in changed:
+        log.log(f"Updated {c}")
+
+
 @shared_task
 def iana_changes_update_task():
     # compensate to avoid we ask for something that happened now and then

ietf/sync/tests.py

@@ -886,6 +886,36 @@ def json(self):
         tasks.rfc_editor_index_update_task(full_index=False)
         self.assertFalse(update_docs_mock.called)
 
+    @override_settings(RFC_EDITOR_QUEUE_URL="https://rfc-editor.example.com/queue/")
+    @mock.patch("ietf.sync.tasks.update_drafts_from_queue")
+    @mock.patch("ietf.sync.tasks.parse_queue")
+    def test_rfc_editor_queue_updates_task(self, mock_parse, mock_update):
+        # test a request timeout
+        self.requests_mock.get("https://rfc-editor.example.com/queue/", exc=requests.exceptions.Timeout)
+        tasks.rfc_editor_queue_updates_task()
+        self.assertFalse(mock_parse.called)
+        self.assertFalse(mock_update.called)
+
+        # now return a value rather than an exception
+        self.requests_mock.get("https://rfc-editor.example.com/queue/", text="the response")
+
+        # mock returning < MIN_QUEUE_RESULTS values - treated as an error, so no update takes place
+        mock_parse.return_value = ([n for n in range(rfceditor.MIN_QUEUE_RESULTS - 1)], ["a warning"])
+        tasks.rfc_editor_queue_updates_task()
+        self.assertEqual(mock_parse.call_count, 1)
+        self.assertEqual(mock_parse.call_args[0][0].read(), "the response")
+        self.assertFalse(mock_update.called)
+        mock_parse.reset_mock()
+
+        # mock returning +. MIN_QUEUE_RESULTS - should succeed
+        mock_parse.return_value = ([n for n in range(rfceditor.MIN_QUEUE_RESULTS)], ["a warning"])
+        mock_update.return_value = ([1,2,3], ["another warning"])
+        tasks.rfc_editor_queue_updates_task()
+        self.assertEqual(mock_parse.call_count, 1)
+        self.assertEqual(mock_parse.call_args[0][0].read(), "the response")
+        self.assertEqual(mock_update.call_count, 1)
+        self.assertEqual(mock_update.call_args, mock.call([n for n in range(rfceditor.MIN_QUEUE_RESULTS)]))
+
     @override_settings(IANA_SYNC_CHANGES_URL="https://iana.example.com/sync/")
     @mock.patch("ietf.sync.tasks.iana.update_history_with_changes")
     @mock.patch("ietf.sync.tasks.iana.parse_changes_json")

ietf/sync/views.py

@@ -2,7 +2,6 @@
 # -*- coding: utf-8 -*-
 
 import datetime
-import subprocess
 import os
 import json
 
@@ -79,30 +78,18 @@ def notify(request, org, notification):
         raise Http404
 
     if request.method == "POST":
-        def runscript(name):
-            python = os.path.join(os.path.dirname(settings.BASE_DIR), "env", "bin", "python")
-            cmd = [python, os.path.join(SYNC_BIN_PATH, name)]
-            cmdstring = " ".join(cmd)
-            p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-            out, err = p.communicate()
-            out = out.decode('utf-8')
-            err = err.decode('utf-8')
-            if p.returncode:
-                log("Subprocess error %s when running '%s': %s %s" % (p.returncode, cmd, err, out))
-                raise subprocess.CalledProcessError(p.returncode, cmdstring, "\n".join([err, out]))
-
         if notification == "index":
             log("Queuing RFC Editor index sync from notify view POST")
             tasks.rfc_editor_index_update_task.delay()
+        elif notification == "queue":
+            log("Queuing RFC Editor queue sync from notify view POST")
+            tasks.rfc_editor_queue_updates_task.delay()
         elif notification == "changes":
             log("Queuing IANA changes sync from notify view POST")
             tasks.iana_changes_update_task.delay()
         elif notification == "protocols":
             log("Queuing IANA protocols sync from notify view POST")
             tasks.iana_protocols_update_task.delay()
-        elif notification == "queue":
-            log("Running sync script from notify view POST")
-            runscript("rfc-editor-queue-updates")
 
         return HttpResponse("OK", content_type="text/plain; charset=%s"%settings.DEFAULT_CHARSET)