Add sync modules (except search) tests to cluster CI (redis#2850)

* Add modules to cluster ci

* remove async tests

* fix protocol checking

* fix tests

* revert cluster docker change

* skip json 2.6.0 tests

* remove breakpoint

* skip test_get_latest

* skip json.mset

* type hint

* revert type hints

* ilnters

---------

Co-authored-by: Chayim <chayim@users.noreply.github.com>

redis/commands/bf/__init__.py

@@ -1,6 +1,6 @@
 from redis._parsers.helpers import bool_ok
 
-from ..helpers import parse_to_list
+from ..helpers import get_protocol_version, parse_to_list
 from .commands import *  # noqa
 from .info import BFInfo, CFInfo, CMSInfo, TDigestInfo, TopKInfo
 
@@ -108,7 +108,7 @@ def __init__(self, client, **kwargs):
         self.commandmixin = CMSCommands
         self.execute_command = client.execute_command
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             _MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             _MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)
@@ -139,7 +139,7 @@ def __init__(self, client, **kwargs):
         self.commandmixin = TOPKCommands
         self.execute_command = client.execute_command
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             _MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             _MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)
@@ -174,7 +174,7 @@ def __init__(self, client, **kwargs):
         self.commandmixin = CFCommands
         self.execute_command = client.execute_command
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             _MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             _MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)
@@ -210,7 +210,7 @@ def __init__(self, client, **kwargs):
         self.commandmixin = TDigestCommands
         self.execute_command = client.execute_command
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             _MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             _MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)
@@ -244,7 +244,7 @@ def __init__(self, client, **kwargs):
         self.commandmixin = BFCommands
         self.execute_command = client.execute_command
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             _MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             _MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)

redis/commands/helpers.py

@@ -3,6 +3,7 @@
 import string
 from typing import List, Tuple
 
+import redis
 from redis.typing import KeysT, KeyT
 
 
@@ -156,3 +157,10 @@ def stringify_param_value(value):
         return f'{{{",".join(f"{k}:{stringify_param_value(v)}" for k, v in value.items())}}}'  # noqa
     else:
         return str(value)
+
+
+def get_protocol_version(client):
+    if isinstance(client, redis.Redis) or isinstance(client, redis.asyncio.Redis):
+        return client.connection_pool.connection_kwargs.get("protocol")
+    elif isinstance(client, redis.cluster.AbstractRedisCluster):
+        return client.nodes_manager.connection_kwargs.get("protocol")

redis/commands/json/__init__.py

@@ -2,7 +2,7 @@
 
 import redis
 
-from ..helpers import nativestr
+from ..helpers import get_protocol_version, nativestr
 from .commands import JSONCommands
 from .decoders import bulk_of_jsons, decode_list
 
@@ -34,6 +34,7 @@ def __init__(
         self._MODULE_CALLBACKS = {
             "JSON.ARRPOP": self._decode,
             "JSON.DEBUG": self._decode,
+            "JSON.GET": self._decode,
             "JSON.MERGE": lambda r: r and nativestr(r) == "OK",
             "JSON.MGET": bulk_of_jsons(self._decode),
             "JSON.MSET": lambda r: r and nativestr(r) == "OK",
@@ -61,19 +62,13 @@ def __init__(
             "JSON.TOGGLE": self._decode,
         }
 
-        _RESP3_MODULE_CALLBACKS = {
-            "JSON.GET": lambda response: [
-                [self._decode(r) for r in res] for res in response
-            ]
-            if response
-            else response
-        }
+        _RESP3_MODULE_CALLBACKS = {}
 
         self.client = client
         self.execute_command = client.execute_command
         self.MODULE_VERSION = version
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             self._MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             self._MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)

redis/commands/timeseries/__init__.py

@@ -1,7 +1,7 @@
 import redis
 from redis._parsers.helpers import bool_ok
 
-from ..helpers import parse_to_list
+from ..helpers import get_protocol_version, parse_to_list
 from .commands import (
     ALTER_CMD,
     CREATE_CMD,
@@ -56,7 +56,7 @@ def __init__(self, client=None, **kwargs):
         self.client = client
         self.execute_command = client.execute_command
 
-        if self.client.connection_pool.connection_kwargs.get("protocol") in ["3", 3]:
+        if get_protocol_version(self.client) in ["3", 3]:
             self._MODULE_CALLBACKS.update(_RESP3_MODULE_CALLBACKS)
         else:
             self._MODULE_CALLBACKS.update(_RESP2_MODULE_CALLBACKS)

tests/test_asyncio/test_bloom.py

@@ -365,7 +365,7 @@ async def test_tdigest_reset(decoded_r: redis.Redis):
 
 
 @pytest.mark.redismod
-@pytest.mark.experimental
+@pytest.mark.onlynoncluster
 async def test_tdigest_merge(decoded_r: redis.Redis):
     assert await decoded_r.tdigest().create("to-tDigest", 10)
     assert await decoded_r.tdigest().create("from-tDigest", 10)

tests/test_asyncio/test_json.py

@@ -112,6 +112,7 @@ async def test_mgetshouldsucceed(decoded_r: redis.Redis):
 
 
 @pytest.mark.redismod
+@pytest.mark.onlynoncluster
 @skip_ifmodversion_lt("2.6.0", "ReJSON")
 async def test_mset(decoded_r: redis.Redis):
     await decoded_r.json().mset(

tests/test_bloom.py

@@ -24,7 +24,6 @@ def client(decoded_r):
     return decoded_r
 
 
-@pytest.mark.redismod
 def test_create(client):
     """Test CREATE/RESERVE calls"""
     assert client.bf().create("bloom", 0.01, 1000)
@@ -39,7 +38,6 @@ def test_create(client):
     assert client.topk().reserve("topk", 5, 100, 5, 0.9)
 
 
-@pytest.mark.redismod
 def test_bf_reserve(client):
     """Testing BF.RESERVE"""
     assert client.bf().reserve("bloom", 0.01, 1000)
@@ -54,13 +52,11 @@ def test_bf_reserve(client):
     assert client.topk().reserve("topk", 5, 100, 5, 0.9)
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_create(client):
     assert client.tdigest().create("tDigest", 100)
 
 
-@pytest.mark.redismod
 def test_bf_add(client):
     assert client.bf().create("bloom", 0.01, 1000)
     assert 1 == client.bf().add("bloom", "foo")
@@ -73,7 +69,6 @@ def test_bf_add(client):
     assert [1, 0] == intlist(client.bf().mexists("bloom", "foo", "noexist"))
 
 
-@pytest.mark.redismod
 def test_bf_insert(client):
     assert client.bf().create("bloom", 0.01, 1000)
     assert [1] == intlist(client.bf().insert("bloom", ["foo"]))
@@ -104,7 +99,6 @@ def test_bf_insert(client):
     )
 
 
-@pytest.mark.redismod
 def test_bf_scandump_and_loadchunk(client):
     # Store a filter
     client.bf().create("myBloom", "0.0001", "1000")
@@ -156,7 +150,6 @@ def do_verify():
     client.bf().create("myBloom", "0.0001", "10000000")
 
 
-@pytest.mark.redismod
 def test_bf_info(client):
     expansion = 4
     # Store a filter
@@ -188,7 +181,6 @@ def test_bf_info(client):
         assert True
 
 
-@pytest.mark.redismod
 def test_bf_card(client):
     # return 0 if the key does not exist
     assert client.bf().card("not_exist") == 0
@@ -203,7 +195,6 @@ def test_bf_card(client):
         client.bf().card("setKey")
 
 
-@pytest.mark.redismod
 def test_cf_add_and_insert(client):
     assert client.cf().create("cuckoo", 1000)
     assert client.cf().add("cuckoo", "filter")
@@ -229,7 +220,6 @@ def test_cf_add_and_insert(client):
     )
 
 
-@pytest.mark.redismod
 def test_cf_exists_and_del(client):
     assert client.cf().create("cuckoo", 1000)
     assert client.cf().add("cuckoo", "filter")
@@ -242,7 +232,6 @@ def test_cf_exists_and_del(client):
     assert 0 == client.cf().count("cuckoo", "filter")
 
 
-@pytest.mark.redismod
 def test_cms(client):
     assert client.cms().initbydim("dim", 1000, 5)
     assert client.cms().initbyprob("prob", 0.01, 0.01)
@@ -258,7 +247,6 @@ def test_cms(client):
     assert 25 == info["count"]
 
 
-@pytest.mark.redismod
 @pytest.mark.onlynoncluster
 def test_cms_merge(client):
     assert client.cms().initbydim("A", 1000, 5)
@@ -276,7 +264,6 @@ def test_cms_merge(client):
     assert [16, 15, 21] == client.cms().query("C", "foo", "bar", "baz")
 
 
-@pytest.mark.redismod
 def test_topk(client):
     # test list with empty buckets
     assert client.topk().reserve("topk", 3, 50, 4, 0.9)
@@ -356,7 +343,6 @@ def test_topk(client):
     assert 0.9 == round(float(info["decay"]), 1)
 
 
-@pytest.mark.redismod
 def test_topk_incrby(client):
     client.flushdb()
     assert client.topk().reserve("topk", 3, 10, 3, 1)
@@ -370,7 +356,6 @@ def test_topk_incrby(client):
         )
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_reset(client):
     assert client.tdigest().create("tDigest", 10)
@@ -387,8 +372,7 @@ def test_tdigest_reset(client):
     )
 
 
-@pytest.mark.redismod
-@pytest.mark.experimental
+@pytest.mark.onlynoncluster
 def test_tdigest_merge(client):
     assert client.tdigest().create("to-tDigest", 10)
     assert client.tdigest().create("from-tDigest", 10)
@@ -415,7 +399,6 @@ def test_tdigest_merge(client):
     assert 4.0 == client.tdigest().max("to-tDigest")
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_min_and_max(client):
     assert client.tdigest().create("tDigest", 100)
@@ -426,7 +409,6 @@ def test_tdigest_min_and_max(client):
     assert 1 == client.tdigest().min("tDigest")
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 @skip_ifmodversion_lt("2.4.0", "bf")
 def test_tdigest_quantile(client):
@@ -448,7 +430,6 @@ def test_tdigest_quantile(client):
     assert [3.0, 5.0] == client.tdigest().quantile("t-digest", 0.5, 0.8)
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_cdf(client):
     assert client.tdigest().create("tDigest", 100)
@@ -460,7 +441,6 @@ def test_tdigest_cdf(client):
     assert [0.1, 0.9] == [round(x, 1) for x in res]
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 @skip_ifmodversion_lt("2.4.0", "bf")
 def test_tdigest_trimmed_mean(client):
@@ -471,7 +451,6 @@ def test_tdigest_trimmed_mean(client):
     assert 4.5 == client.tdigest().trimmed_mean("tDigest", 0.4, 0.5)
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_rank(client):
     assert client.tdigest().create("t-digest", 500)
@@ -482,7 +461,6 @@ def test_tdigest_rank(client):
     assert [-1, 20, 9] == client.tdigest().rank("t-digest", -20, 20, 9)
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_revrank(client):
     assert client.tdigest().create("t-digest", 500)
@@ -492,7 +470,6 @@ def test_tdigest_revrank(client):
     assert [-1, 19, 9] == client.tdigest().revrank("t-digest", 21, 0, 10)
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_byrank(client):
     assert client.tdigest().create("t-digest", 500)
@@ -504,7 +481,6 @@ def test_tdigest_byrank(client):
         client.tdigest().byrank("t-digest", -1)[0]
 
 
-@pytest.mark.redismod
 @pytest.mark.experimental
 def test_tdigest_byrevrank(client):
     assert client.tdigest().create("t-digest", 500)
@@ -516,8 +492,7 @@ def test_tdigest_byrevrank(client):
         client.tdigest().byrevrank("t-digest", -1)[0]
 
 
-# @pytest.mark.redismod
-# def test_pipeline(client):
+# # def test_pipeline(client):
 #     pipeline = client.bf().pipeline()
 #     assert not client.bf().execute_command("get pipeline")
 #