feat(celery): VALKEY_SCHEME environment variable (#10420)

Co-authored-by: Pepe Fagoaga <pepe@prowler.com>

api/src/backend/api/tests/test_celery_settings.py

@@ -0,0 +1,43 @@
+import pytest
+from config.settings.celery import _build_celery_broker_url
+
+
+class TestBuildCeleryBrokerUrl:
+    def test_without_credentials(self):
+        broker_url = _build_celery_broker_url("redis", "", "", "valkey", "6379", "0")
+
+        assert broker_url == "redis://valkey:6379/0"
+
+    def test_with_password_only(self):
+        broker_url = _build_celery_broker_url(
+            "rediss", "", "secret", "cache.example.com", "6379", "0"
+        )
+
+        assert broker_url == "rediss://:secret@cache.example.com:6379/0"
+
+    def test_with_username_and_password(self):
+        broker_url = _build_celery_broker_url(
+            "rediss", "default", "secret", "cache.example.com", "6379", "0"
+        )
+
+        assert broker_url == "rediss://default:secret@cache.example.com:6379/0"
+
+    def test_with_username_only(self):
+        broker_url = _build_celery_broker_url(
+            "redis", "admin", "", "valkey", "6379", "0"
+        )
+
+        assert broker_url == "redis://admin@valkey:6379/0"
+
+    def test_url_encodes_credentials(self):
+        broker_url = _build_celery_broker_url(
+            "rediss", "user@name", "p@ss:word", "cache.example.com", "6379", "0"
+        )
+
+        assert (
+            broker_url == "rediss://user%40name:p%40ss%3Aword@cache.example.com:6379/0"
+        )
+
+    def test_invalid_scheme_raises_error(self):
+        with pytest.raises(ValueError, match="Invalid VALKEY_SCHEME 'http'"):
+            _build_celery_broker_url("http", "", "", "valkey", "6379", "0")

api/src/backend/config/settings/celery.py

@@ -1,10 +1,52 @@
+from urllib.parse import quote
+
 from config.env import env
 
+_VALID_SCHEMES = {"redis", "rediss"}
+
+
+def _build_celery_broker_url(
+    scheme: str,
+    username: str,
+    password: str,
+    host: str,
+    port: str,
+    db: str,
+) -> str:
+    if scheme not in _VALID_SCHEMES:
+        raise ValueError(
+            f"Invalid VALKEY_SCHEME '{scheme}'. Must be one of: {', '.join(sorted(_VALID_SCHEMES))}"
+        )
+
+    encoded_username = quote(username, safe="") if username else ""
+    encoded_password = quote(password, safe="") if password else ""
+
+    auth = ""
+    if encoded_username and encoded_password:
+        auth = f"{encoded_username}:{encoded_password}@"
+    elif encoded_password:
+        auth = f":{encoded_password}@"
+    elif encoded_username:
+        auth = f"{encoded_username}@"
+
+    return f"{scheme}://{auth}{host}:{port}/{db}"
+
+
+VALKEY_SCHEME = env("VALKEY_SCHEME", default="redis")
+VALKEY_USERNAME = env("VALKEY_USERNAME", default="")
+VALKEY_PASSWORD = env("VALKEY_PASSWORD", default="")
 VALKEY_HOST = env("VALKEY_HOST", default="valkey")
 VALKEY_PORT = env("VALKEY_PORT", default="6379")
 VALKEY_DB = env("VALKEY_DB", default="0")
 
-CELERY_BROKER_URL = f"redis://{VALKEY_HOST}:{VALKEY_PORT}/{VALKEY_DB}"
+CELERY_BROKER_URL = _build_celery_broker_url(
+    VALKEY_SCHEME,
+    VALKEY_USERNAME,
+    VALKEY_PASSWORD,
+    VALKEY_HOST,
+    VALKEY_PORT,
+    VALKEY_DB,
+)
 CELERY_RESULT_BACKEND = "django-db"
 CELERY_TASK_TRACK_STARTED = True