fix: python3.12+ compatibilty and redis credential support

python/ciqueue/_pytest/test_queue.py

@@ -1,11 +1,18 @@
-from distutils import util  # pylint: disable=no-name-in-module, import-modules-only
 from future.moves.urllib import parse as urlparse
 import ciqueue
 import ciqueue.distributed
 import redis
 import uritools
 
 
+def strtobool(val):
+    if val.lower() in ('y', 'yes', 't', 'true', 'on', '1'):
+        return True
+    if val.lower() in ('n', 'no', 'f', 'false', 'off', '0'):
+        return False
+    raise ValueError(f"invalid truth value {val!r}")
+
+
 class InvalidRedisUrl(Exception):
     pass
 
@@ -41,17 +48,26 @@ def parse_worker_args(query_string, tests_index):
 def parse_redis_args(spec):
     query = urlparse.parse_qs(spec.query)
 
-    result = {'host': spec.authority.split(':')[0],
+    result = {'host': spec.host,
               'db': int(spec.path[1:] or 0)}
 
+    if spec.userinfo:
+        parts = spec.userinfo.split(':', 1)
+        if len(parts) == 2:
+            if parts[0]:
+                result['username'] = parts[0]
+            result['password'] = parts[1]
+        else:
+            result['password'] = parts[0]
+
     if spec.port:
         result['port'] = spec.port
     if 'socket_timeout' in query:
         result['socket_timeout'] = int(query['socket_timeout'][0])
     if 'socket_connect_timeout' in query:
         result['socket_connect_timeout'] = int(query['socket_connect_timeout'][0])
     if 'retry_on_timeout' in query:
-        result['retry_on_timeout'] = bool(util.strtobool(query['retry_on_timeout'][0] or 'false'))
+        result['retry_on_timeout'] = strtobool(query['retry_on_timeout'][0] or 'false')
     if spec.scheme == "rediss":
         result['ssl'] = True
 

python/tests/test_test_queue.py

@@ -1,3 +1,4 @@
+import uritools
 import ciqueue.distributed
 from ciqueue._pytest import test_queue
 
@@ -12,3 +13,22 @@ def test_initialise_from_rediss_uri(self):
         queue = test_queue.build_queue('rediss://localhost:6379/0?worker=1&build=12345', None)
         assert isinstance(queue, ciqueue.distributed.Supervisor)
         assert queue.redis is not None
+
+    def test_parse_redis_args_with_username_and_password(self):
+        spec = uritools.urisplit('redis://user:secret@localhost:6379/0')
+        args = test_queue.parse_redis_args(spec)
+        assert args['username'] == 'user'
+        assert args['password'] == 'secret'
+        assert args['host'] == 'localhost'
+
+    def test_parse_redis_args_with_password_only(self):
+        spec = uritools.urisplit('redis://:secret@localhost:6379/0')
+        args = test_queue.parse_redis_args(spec)
+        assert 'username' not in args
+        assert args['password'] == 'secret'
+
+    def test_parse_redis_args_without_credentials(self):
+        spec = uritools.urisplit('redis://localhost:6379/0')
+        args = test_queue.parse_redis_args(spec)
+        assert 'username' not in args
+        assert 'password' not in args