feature(huey): use RedisHuey

selfprivacy_api/task_registry.py

@@ -3,3 +3,4 @@ from selfprivacy_api.jobs.test import test_job
 from selfprivacy_api.backup.tasks import *
 from selfprivacy_api.services.tasks import move_service
 from selfprivacy_api.jobs.upgrade_system import rebuild_system_task
+from tests.test_huey import sum

selfprivacy_api/utils/huey.py

@@ -1,16 +1,23 @@
 """MiniHuey singleton."""
 import os
-from huey import SqliteHuey
+from os import environ
+from huey import RedisHuey
 
-HUEY_DATABASE = "/etc/selfprivacy/tasks.db"
+from selfprivacy_api.utils.redis_pool import RedisPool
+
+HUEY_DATABASE_NUMBER = 10
+
+def immediate() -> bool:
+    if environ.get("HUEY_QUEUES_FOR_TESTS"):
+        return False
+    if environ.get("TEST_MODE"):
+        return True
+    return False
 
 # Singleton instance containing the huey database.
-
+huey = RedisHuey(
-test_mode = os.environ.get("TEST_MODE")
-
-huey = SqliteHuey(
     "selfprivacy-api",
-    filename=HUEY_DATABASE if not test_mode else None,
+    url=RedisPool.connection_url(dbnumber=HUEY_DATABASE_NUMBER),
-    immediate=test_mode == "true",
+    immediate=immediate(),
     utc=True,
 )

selfprivacy_api/utils/redis_pool.py

@@ -14,20 +14,25 @@ class RedisPool(metaclass=SingletonMetaclass):
     """
 
     def __init__(self):
-        if "USE_REDIS_PORT" in environ:
+        self._pool = redis.ConnectionPool.from_url(
-            self._pool = redis.ConnectionPool(
+            RedisPool.connection_url(dbnumber=0),
-                host="127.0.0.1",
+            decode_responses=True,
-                port=int(environ["USE_REDIS_PORT"]),
+        )
-                decode_responses=True,
-            )
-
-        else:
-            self._pool = redis.ConnectionPool.from_url(
-                f"unix://{REDIS_SOCKET}",
-                decode_responses=True,
-            )
         self._pubsub_connection = self.get_connection()
 
+    @staticmethod
+    def connection_url(dbnumber: int) -> str:
+        """
+        redis://[[username]:[password]]@localhost:6379/0
+        unix://[username@]/path/to/socket.sock?db=0[&password=password]
+        """
+
+        if "USE_REDIS_PORT" in environ:
+            port = int(environ["USE_REDIS_PORT"])
+            return f"redis://@127.0.0.1:{port}/{dbnumber}"
+        else:
+            return f"unix://{REDIS_SOCKET}?db={dbnumber}"
+
     def get_connection(self):
         """
         Get a connection from the pool.

tests/conftest.py

@@ -99,23 +99,14 @@ def generic_userdata(mocker, tmpdir):
 
 
 @pytest.fixture
-def huey_database(mocker, shared_datadir):
+def client(redis_repo_with_tokens):
-    """Mock huey database."""
-    mock = mocker.patch(
-        "selfprivacy_api.utils.huey.HUEY_DATABASE", shared_datadir / "huey.db"
-    )
-    return mock
-
-
-@pytest.fixture
-def client(huey_database, redis_repo_with_tokens):
     from selfprivacy_api.app import app
 
     return TestClient(app)
 
 
 @pytest.fixture
-def authorized_client(huey_database, redis_repo_with_tokens):
+def authorized_client(redis_repo_with_tokens):
     """Authorized test client fixture."""
     from selfprivacy_api.app import app
 
@@ -127,7 +118,7 @@ def authorized_client(huey_database, redis_repo_with_tokens):
 
 
 @pytest.fixture
-def wrong_auth_client(huey_database, redis_repo_with_tokens):
+def wrong_auth_client(redis_repo_with_tokens):
     """Wrong token test client fixture."""
     from selfprivacy_api.app import app
 

tests/test_huey.py

@@ -0,0 +1,41 @@
+import pytest
+
+from subprocess import Popen
+from os import environ
+
+
+# from selfprivacy_api.backup.util import output_yielder
+from selfprivacy_api.utils.huey import huey
+
+
+@huey.task()
+def sum(a: int, b: int) -> int:
+    return a + b
+
+
+@pytest.fixture()
+def huey_queues():
+    """
+    Full, not-immediate, queued huey, with consumer starting and stopping.
+    IMPORTANT: Assumes tests are run from the project directory.
+    The above is needed by consumer to find our huey setup.
+    """
+    old_immediate = huey.immediate
+
+    environ["HUEY_QUEUES_FOR_TESTS"] = "Yes"
+    command = ["huey_consumer.py", "selfprivacy_api.task_registry.huey"]
+    huey.immediate = False
+    assert huey.immediate is False
+    consumer_handle = Popen(command)
+
+    yield huey
+
+    consumer_handle.terminate()
+    del environ["HUEY_QUEUES_FOR_TESTS"]
+    huey.immediate = old_immediate
+    assert huey.immediate == old_immediate
+
+
+def test_huey(huey_queues):
+    result = sum(2, 5)
+    assert result(blocking=True) == 7