From 9db1eec71b443b8e7e74ff503bae651dc6edf411 Mon Sep 17 00:00:00 2001 From: Bar Shaul <88437685+barshaul@users.noreply.github.com> Date: Thu, 25 Nov 2021 14:15:24 +0200 Subject: Adding RedisCluster client to support Redis Cluster Mode (#1660) Co-authored-by: Chayim Co-authored-by: Anas --- tests/test_cluster.py | 2482 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 2482 insertions(+) create mode 100644 tests/test_cluster.py (limited to 'tests/test_cluster.py') diff --git a/tests/test_cluster.py b/tests/test_cluster.py new file mode 100644 index 0000000..071cb7d --- /dev/null +++ b/tests/test_cluster.py @@ -0,0 +1,2482 @@ +import binascii +import datetime +import pytest +import warnings + +from time import sleep +from tests.test_pubsub import wait_for_message +from unittest.mock import call, patch, DEFAULT, Mock +from redis import Redis +from redis.cluster import get_node_name, ClusterNode, \ + RedisCluster, NodesManager, PRIMARY, REDIS_CLUSTER_HASH_SLOTS, REPLICA +from redis.commands import CommandsParser +from redis.connection import Connection +from redis.utils import str_if_bytes +from redis.exceptions import ( + AskError, + ClusterDownError, + DataError, + MovedError, + RedisClusterException, + RedisError +) + +from redis.crc import key_slot +from .conftest import ( + _get_client, + skip_if_server_version_lt, + skip_unless_arch_bits +) + +default_host = "127.0.0.1" +default_port = 7000 +default_cluster_slots = [ + [ + 0, 8191, + ['127.0.0.1', 7000, 'node_0'], + ['127.0.0.1', 7003, 'node_3'], + ], + [ + 8192, 16383, + ['127.0.0.1', 7001, 'node_1'], + ['127.0.0.1', 7002, 'node_2'] + ] +] + + +@pytest.fixture() +def slowlog(request, r): + """ + Set the slowlog threshold to 0, and the + max length to 128. This will force every + command into the slowlog and allow us + to test it + """ + # Save old values + current_config = r.config_get( + target_nodes=r.get_primaries()[0]) + old_slower_than_value = current_config['slowlog-log-slower-than'] + old_max_legnth_value = current_config['slowlog-max-len'] + + # Function to restore the old values + def cleanup(): + r.config_set('slowlog-log-slower-than', old_slower_than_value) + r.config_set('slowlog-max-len', old_max_legnth_value) + + request.addfinalizer(cleanup) + + # Set the new values + r.config_set('slowlog-log-slower-than', 0) + r.config_set('slowlog-max-len', 128) + + +def get_mocked_redis_client(func=None, *args, **kwargs): + """ + Return a stable RedisCluster object that have deterministic + nodes and slots setup to remove the problem of different IP addresses + on different installations and machines. + """ + cluster_slots = kwargs.pop('cluster_slots', default_cluster_slots) + coverage_res = kwargs.pop('coverage_result', 'yes') + with patch.object(Redis, 'execute_command') as execute_command_mock: + def execute_command(*_args, **_kwargs): + if _args[0] == 'CLUSTER SLOTS': + mock_cluster_slots = cluster_slots + return mock_cluster_slots + elif _args[0] == 'COMMAND': + return {'get': [], 'set': []} + elif _args[1] == 'cluster-require-full-coverage': + return {'cluster-require-full-coverage': coverage_res} + elif func is not None: + return func(*args, **kwargs) + else: + return execute_command_mock(*_args, **_kwargs) + + execute_command_mock.side_effect = execute_command + + with patch.object(CommandsParser, 'initialize', + autospec=True) as cmd_parser_initialize: + + def cmd_init_mock(self, r): + self.commands = {'get': {'name': 'get', 'arity': 2, + 'flags': ['readonly', + 'fast'], + 'first_key_pos': 1, + 'last_key_pos': 1, + 'step_count': 1}} + + cmd_parser_initialize.side_effect = cmd_init_mock + + return RedisCluster(*args, **kwargs) + + +def mock_node_resp(node, response): + connection = Mock() + connection.read_response.return_value = response + node.redis_connection.connection = connection + return node + + +def mock_node_resp_func(node, func): + connection = Mock() + connection.read_response.side_effect = func + node.redis_connection.connection = connection + return node + + +def mock_all_nodes_resp(rc, response): + for node in rc.get_nodes(): + mock_node_resp(node, response) + return rc + + +def find_node_ip_based_on_port(cluster_client, port): + for node in cluster_client.get_nodes(): + if node.port == port: + return node.host + + +def moved_redirection_helper(request, failover=False): + """ + Test that the client handles MOVED response after a failover. + Redirection after a failover means that the redirection address is of a + replica that was promoted to a primary. + + At first call it should return a MOVED ResponseError that will point + the client to the next server it should talk to. + + Verify that: + 1. it tries to talk to the redirected node + 2. it updates the slot's primary to the redirected node + + For a failover, also verify: + 3. the redirected node's server type updated to 'primary' + 4. the server type of the previous slot owner updated to 'replica' + """ + rc = _get_client(RedisCluster, request, flushdb=False) + slot = 12182 + redirect_node = None + # Get the current primary that holds this slot + prev_primary = rc.nodes_manager.get_node_from_slot(slot) + if failover: + if len(rc.nodes_manager.slots_cache[slot]) < 2: + warnings.warn("Skipping this test since it requires to have a " + "replica") + return + redirect_node = rc.nodes_manager.slots_cache[slot][1] + else: + # Use one of the primaries to be the redirected node + redirect_node = rc.get_primaries()[0] + r_host = redirect_node.host + r_port = redirect_node.port + with patch.object(Redis, 'parse_response') as parse_response: + def moved_redirect_effect(connection, *args, **options): + def ok_response(connection, *args, **options): + assert connection.host == r_host + assert connection.port == r_port + + return "MOCK_OK" + + parse_response.side_effect = ok_response + raise MovedError("{0} {1}:{2}".format(slot, r_host, r_port)) + + parse_response.side_effect = moved_redirect_effect + assert rc.execute_command("SET", "foo", "bar") == "MOCK_OK" + slot_primary = rc.nodes_manager.slots_cache[slot][0] + assert slot_primary == redirect_node + if failover: + assert rc.get_node(host=r_host, port=r_port).server_type == PRIMARY + assert prev_primary.server_type == REPLICA + + +@pytest.mark.onlycluster +class TestRedisClusterObj: + """ + Tests for the RedisCluster class + """ + + def test_host_port_startup_node(self): + """ + Test that it is possible to use host & port arguments as startup node + args + """ + cluster = get_mocked_redis_client(host=default_host, port=default_port) + assert cluster.get_node(host=default_host, + port=default_port) is not None + + def test_startup_nodes(self): + """ + Test that it is possible to use startup_nodes + argument to init the cluster + """ + port_1 = 7000 + port_2 = 7001 + startup_nodes = [ClusterNode(default_host, port_1), + ClusterNode(default_host, port_2)] + cluster = get_mocked_redis_client(startup_nodes=startup_nodes) + assert cluster.get_node(host=default_host, port=port_1) is not None \ + and cluster.get_node(host=default_host, port=port_2) is not None + + def test_empty_startup_nodes(self): + """ + Test that exception is raised when empty providing empty startup_nodes + """ + with pytest.raises(RedisClusterException) as ex: + RedisCluster(startup_nodes=[]) + + assert str(ex.value).startswith( + "RedisCluster requires at least one node to discover the " + "cluster"), str_if_bytes(ex.value) + + def test_from_url(self, r): + redis_url = "redis://{0}:{1}/0".format(default_host, default_port) + with patch.object(RedisCluster, 'from_url') as from_url: + def from_url_mocked(_url, **_kwargs): + return get_mocked_redis_client(url=_url, **_kwargs) + + from_url.side_effect = from_url_mocked + cluster = RedisCluster.from_url(redis_url) + assert cluster.get_node(host=default_host, + port=default_port) is not None + + def test_execute_command_errors(self, r): + """ + Test that if no key is provided then exception should be raised. + """ + with pytest.raises(RedisClusterException) as ex: + r.execute_command("GET") + assert str(ex.value).startswith("No way to dispatch this command to " + "Redis Cluster. Missing key.") + + def test_execute_command_node_flag_primaries(self, r): + """ + Test command execution with nodes flag PRIMARIES + """ + primaries = r.get_primaries() + replicas = r.get_replicas() + mock_all_nodes_resp(r, 'PONG') + assert r.ping(RedisCluster.PRIMARIES) is True + for primary in primaries: + conn = primary.redis_connection.connection + assert conn.read_response.called is True + for replica in replicas: + conn = replica.redis_connection.connection + assert conn.read_response.called is not True + + def test_execute_command_node_flag_replicas(self, r): + """ + Test command execution with nodes flag REPLICAS + """ + replicas = r.get_replicas() + if not replicas: + r = get_mocked_redis_client(default_host, default_port) + primaries = r.get_primaries() + mock_all_nodes_resp(r, 'PONG') + assert r.ping(RedisCluster.REPLICAS) is True + for replica in replicas: + conn = replica.redis_connection.connection + assert conn.read_response.called is True + for primary in primaries: + conn = primary.redis_connection.connection + assert conn.read_response.called is not True + + def test_execute_command_node_flag_all_nodes(self, r): + """ + Test command execution with nodes flag ALL_NODES + """ + mock_all_nodes_resp(r, 'PONG') + assert r.ping(RedisCluster.ALL_NODES) is True + for node in r.get_nodes(): + conn = node.redis_connection.connection + assert conn.read_response.called is True + + def test_execute_command_node_flag_random(self, r): + """ + Test command execution with nodes flag RANDOM + """ + mock_all_nodes_resp(r, 'PONG') + assert r.ping(RedisCluster.RANDOM) is True + called_count = 0 + for node in r.get_nodes(): + conn = node.redis_connection.connection + if conn.read_response.called is True: + called_count += 1 + assert called_count == 1 + + def test_execute_command_default_node(self, r): + """ + Test command execution without node flag is being executed on the + default node + """ + def_node = r.get_default_node() + mock_node_resp(def_node, 'PONG') + assert r.ping() is True + conn = def_node.redis_connection.connection + assert conn.read_response.called + + def test_ask_redirection(self, r): + """ + Test that the server handles ASK response. + + At first call it should return a ASK ResponseError that will point + the client to the next server it should talk to. + + Important thing to verify is that it tries to talk to the second node. + """ + redirect_node = r.get_nodes()[0] + with patch.object(Redis, 'parse_response') as parse_response: + def ask_redirect_effect(connection, *args, **options): + def ok_response(connection, *args, **options): + assert connection.host == redirect_node.host + assert connection.port == redirect_node.port + + return "MOCK_OK" + + parse_response.side_effect = ok_response + raise AskError("12182 {0}:{1}".format(redirect_node.host, + redirect_node.port)) + + parse_response.side_effect = ask_redirect_effect + + assert r.execute_command("SET", "foo", "bar") == "MOCK_OK" + + def test_moved_redirection(self, request): + """ + Test that the client handles MOVED response. + """ + moved_redirection_helper(request, failover=False) + + def test_moved_redirection_after_failover(self, request): + """ + Test that the client handles MOVED response after a failover. + """ + moved_redirection_helper(request, failover=True) + + def test_refresh_using_specific_nodes(self, request): + """ + Test making calls on specific nodes when the cluster has failed over to + another node + """ + node_7006 = ClusterNode(host=default_host, port=7006, + server_type=PRIMARY) + node_7007 = ClusterNode(host=default_host, port=7007, + server_type=PRIMARY) + with patch.object(Redis, 'parse_response') as parse_response: + with patch.object(NodesManager, 'initialize', autospec=True) as \ + initialize: + with patch.multiple(Connection, + send_command=DEFAULT, + connect=DEFAULT, + can_read=DEFAULT) as mocks: + # simulate 7006 as a failed node + def parse_response_mock(connection, command_name, + **options): + if connection.port == 7006: + parse_response.failed_calls += 1 + raise ClusterDownError( + 'CLUSTERDOWN The cluster is ' + 'down. Use CLUSTER INFO for ' + 'more information') + elif connection.port == 7007: + parse_response.successful_calls += 1 + + def initialize_mock(self): + # start with all slots mapped to 7006 + self.nodes_cache = {node_7006.name: node_7006} + self.default_node = node_7006 + self.slots_cache = {} + + for i in range(0, 16383): + self.slots_cache[i] = [node_7006] + + # After the first connection fails, a reinitialize + # should follow the cluster to 7007 + def map_7007(self): + self.nodes_cache = { + node_7007.name: node_7007} + self.default_node = node_7007 + self.slots_cache = {} + + for i in range(0, 16383): + self.slots_cache[i] = [node_7007] + + # Change initialize side effect for the second call + initialize.side_effect = map_7007 + + parse_response.side_effect = parse_response_mock + parse_response.successful_calls = 0 + parse_response.failed_calls = 0 + initialize.side_effect = initialize_mock + mocks['can_read'].return_value = False + mocks['send_command'].return_value = "MOCK_OK" + mocks['connect'].return_value = None + with patch.object(CommandsParser, 'initialize', + autospec=True) as cmd_parser_initialize: + + def cmd_init_mock(self, r): + self.commands = {'get': {'name': 'get', 'arity': 2, + 'flags': ['readonly', + 'fast'], + 'first_key_pos': 1, + 'last_key_pos': 1, + 'step_count': 1}} + + cmd_parser_initialize.side_effect = cmd_init_mock + + rc = _get_client( + RedisCluster, request, flushdb=False) + assert len(rc.get_nodes()) == 1 + assert rc.get_node(node_name=node_7006.name) is not \ + None + + rc.get('foo') + + # Cluster should now point to 7007, and there should be + # one failed and one successful call + assert len(rc.get_nodes()) == 1 + assert rc.get_node(node_name=node_7007.name) is not \ + None + assert rc.get_node(node_name=node_7006.name) is None + assert parse_response.failed_calls == 1 + assert parse_response.successful_calls == 1 + + def test_reading_from_replicas_in_round_robin(self): + with patch.multiple(Connection, send_command=DEFAULT, + read_response=DEFAULT, _connect=DEFAULT, + can_read=DEFAULT, on_connect=DEFAULT) as mocks: + with patch.object(Redis, 'parse_response') as parse_response: + def parse_response_mock_first(connection, *args, **options): + # Primary + assert connection.port == 7001 + parse_response.side_effect = parse_response_mock_second + return "MOCK_OK" + + def parse_response_mock_second(connection, *args, **options): + # Replica + assert connection.port == 7002 + parse_response.side_effect = parse_response_mock_third + return "MOCK_OK" + + def parse_response_mock_third(connection, *args, **options): + # Primary + assert connection.port == 7001 + return "MOCK_OK" + + # We don't need to create a real cluster connection but we + # do want RedisCluster.on_connect function to get called, + # so we'll mock some of the Connection's functions to allow it + parse_response.side_effect = parse_response_mock_first + mocks['send_command'].return_value = True + mocks['read_response'].return_value = "OK" + mocks['_connect'].return_value = True + mocks['can_read'].return_value = False + mocks['on_connect'].return_value = True + + # Create a cluster with reading from replications + read_cluster = get_mocked_redis_client(host=default_host, + port=default_port, + read_from_replicas=True) + assert read_cluster.read_from_replicas is True + # Check that we read from the slot's nodes in a round robin + # matter. + # 'foo' belongs to slot 12182 and the slot's nodes are: + # [(127.0.0.1,7001,primary), (127.0.0.1,7002,replica)] + read_cluster.get("foo") + read_cluster.get("foo") + read_cluster.get("foo") + mocks['send_command'].assert_has_calls([call('READONLY')]) + + def test_keyslot(self, r): + """ + Test that method will compute correct key in all supported cases + """ + assert r.keyslot("foo") == 12182 + assert r.keyslot("{foo}bar") == 12182 + assert r.keyslot("{foo}") == 12182 + assert r.keyslot(1337) == 4314 + + assert r.keyslot(125) == r.keyslot(b"125") + assert r.keyslot(125) == r.keyslot("\x31\x32\x35") + assert r.keyslot("大奖") == r.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96") + assert r.keyslot(u"大奖") == r.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96") + assert r.keyslot(1337.1234) == r.keyslot("1337.1234") + assert r.keyslot(1337) == r.keyslot("1337") + assert r.keyslot(b"abc") == r.keyslot("abc") + + def test_get_node_name(self): + assert get_node_name(default_host, default_port) == \ + "{0}:{1}".format(default_host, default_port) + + def test_all_nodes(self, r): + """ + Set a list of nodes and it should be possible to iterate over all + """ + nodes = [node for node in r.nodes_manager.nodes_cache.values()] + + for i, node in enumerate(r.get_nodes()): + assert node in nodes + + def test_all_nodes_masters(self, r): + """ + Set a list of nodes with random primaries/replicas config and it shold + be possible to iterate over all of them. + """ + nodes = [node for node in r.nodes_manager.nodes_cache.values() + if node.server_type == PRIMARY] + + for node in r.get_primaries(): + assert node in nodes + + def test_cluster_down_overreaches_retry_attempts(self): + """ + When ClusterDownError is thrown, test that we retry executing the + command as many times as configured in cluster_error_retry_attempts + and then raise the exception + """ + with patch.object(RedisCluster, '_execute_command') as execute_command: + def raise_cluster_down_error(target_node, *args, **kwargs): + execute_command.failed_calls += 1 + raise ClusterDownError( + 'CLUSTERDOWN The cluster is down. Use CLUSTER INFO for ' + 'more information') + + execute_command.side_effect = raise_cluster_down_error + + rc = get_mocked_redis_client(host=default_host, port=default_port) + + with pytest.raises(ClusterDownError): + rc.get("bar") + assert execute_command.failed_calls == \ + rc.cluster_error_retry_attempts + + def test_connection_error_overreaches_retry_attempts(self): + """ + When ConnectionError is thrown, test that we retry executing the + command as many times as configured in cluster_error_retry_attempts + and then raise the exception + """ + with patch.object(RedisCluster, '_execute_command') as execute_command: + def raise_conn_error(target_node, *args, **kwargs): + execute_command.failed_calls += 1 + raise ConnectionError() + + execute_command.side_effect = raise_conn_error + + rc = get_mocked_redis_client(host=default_host, port=default_port) + + with pytest.raises(ConnectionError): + rc.get("bar") + assert execute_command.failed_calls == \ + rc.cluster_error_retry_attempts + + def test_user_on_connect_function(self, request): + """ + Test support in passing on_connect function by the user + """ + + def on_connect(connection): + assert connection is not None + + mock = Mock(side_effect=on_connect) + + _get_client(RedisCluster, request, redis_connect_func=mock) + assert mock.called is True + + def test_set_default_node_success(self, r): + """ + test successful replacement of the default cluster node + """ + default_node = r.get_default_node() + # get a different node + new_def_node = None + for node in r.get_nodes(): + if node != default_node: + new_def_node = node + break + assert r.set_default_node(new_def_node) is True + assert r.get_default_node() == new_def_node + + def test_set_default_node_failure(self, r): + """ + test failed replacement of the default cluster node + """ + default_node = r.get_default_node() + new_def_node = ClusterNode('1.1.1.1', 1111) + assert r.set_default_node(None) is False + assert r.set_default_node(new_def_node) is False + assert r.get_default_node() == default_node + + def test_get_node_from_key(self, r): + """ + Test that get_node_from_key function returns the correct node + """ + key = 'bar' + slot = r.keyslot(key) + slot_nodes = r.nodes_manager.slots_cache.get(slot) + primary = slot_nodes[0] + assert r.get_node_from_key(key, replica=False) == primary + replica = r.get_node_from_key(key, replica=True) + if replica is not None: + assert replica.server_type == REPLICA + assert replica in slot_nodes + + +@pytest.mark.onlycluster +class TestClusterRedisCommands: + """ + Tests for RedisCluster unique commands + """ + + def test_case_insensitive_command_names(self, r): + assert r.cluster_response_callbacks['cluster addslots'] == \ + r.cluster_response_callbacks['CLUSTER ADDSLOTS'] + + def test_get_and_set(self, r): + # get and set can't be tested independently of each other + assert r.get('a') is None + byte_string = b'value' + integer = 5 + unicode_string = chr(3456) + 'abcd' + chr(3421) + assert r.set('byte_string', byte_string) + assert r.set('integer', 5) + assert r.set('unicode_string', unicode_string) + assert r.get('byte_string') == byte_string + assert r.get('integer') == str(integer).encode() + assert r.get('unicode_string').decode('utf-8') == unicode_string + + def test_mget_nonatomic(self, r): + assert r.mget_nonatomic([]) == [] + assert r.mget_nonatomic(['a', 'b']) == [None, None] + r['a'] = '1' + r['b'] = '2' + r['c'] = '3' + + assert (r.mget_nonatomic('a', 'other', 'b', 'c') == + [b'1', None, b'2', b'3']) + + def test_mset_nonatomic(self, r): + d = {'a': b'1', 'b': b'2', 'c': b'3', 'd': b'4'} + assert r.mset_nonatomic(d) + for k, v in d.items(): + assert r[k] == v + + def test_config_set(self, r): + assert r.config_set('slowlog-log-slower-than', 0) + + def test_cluster_config_resetstat(self, r): + r.ping(target_nodes='all') + all_info = r.info(target_nodes='all') + prior_commands_processed = -1 + for node_info in all_info.values(): + prior_commands_processed = node_info['total_commands_processed'] + assert prior_commands_processed >= 1 + r.config_resetstat(target_nodes='all') + all_info = r.info(target_nodes='all') + for node_info in all_info.values(): + reset_commands_processed = node_info['total_commands_processed'] + assert reset_commands_processed < prior_commands_processed + + def test_client_setname(self, r): + node = r.get_random_node() + r.client_setname('redis_py_test', target_nodes=node) + client_name = r.client_getname(target_nodes=node) + assert client_name == 'redis_py_test' + + def test_exists(self, r): + d = {'a': b'1', 'b': b'2', 'c': b'3', 'd': b'4'} + r.mset_nonatomic(d) + assert r.exists(*d.keys()) == len(d) + + def test_delete(self, r): + d = {'a': b'1', 'b': b'2', 'c': b'3', 'd': b'4'} + r.mset_nonatomic(d) + assert r.delete(*d.keys()) == len(d) + assert r.delete(*d.keys()) == 0 + + def test_touch(self, r): + d = {'a': b'1', 'b': b'2', 'c': b'3', 'd': b'4'} + r.mset_nonatomic(d) + assert r.touch(*d.keys()) == len(d) + + def test_unlink(self, r): + d = {'a': b'1', 'b': b'2', 'c': b'3', 'd': b'4'} + r.mset_nonatomic(d) + assert r.unlink(*d.keys()) == len(d) + # Unlink is non-blocking so we sleep before + # verifying the deletion + sleep(0.1) + assert r.unlink(*d.keys()) == 0 + + def test_pubsub_channels_merge_results(self, r): + nodes = r.get_nodes() + channels = [] + pubsub_nodes = [] + i = 0 + for node in nodes: + channel = "foo{0}".format(i) + # We will create different pubsub clients where each one is + # connected to a different node + p = r.pubsub(node) + pubsub_nodes.append(p) + p.subscribe(channel) + b_channel = channel.encode('utf-8') + channels.append(b_channel) + # Assert that each node returns only the channel it subscribed to + sub_channels = node.redis_connection.pubsub_channels() + if not sub_channels: + # Try again after a short sleep + sleep(0.3) + sub_channels = node.redis_connection.pubsub_channels() + assert sub_channels == [b_channel] + i += 1 + # Assert that the cluster's pubsub_channels function returns ALL of + # the cluster's channels + result = r.pubsub_channels(target_nodes='all') + result.sort() + assert result == channels + + def test_pubsub_numsub_merge_results(self, r): + nodes = r.get_nodes() + pubsub_nodes = [] + channel = "foo" + b_channel = channel.encode('utf-8') + for node in nodes: + # We will create different pubsub clients where each one is + # connected to a different node + p = r.pubsub(node) + pubsub_nodes.append(p) + p.subscribe(channel) + # Assert that each node returns that only one client is subscribed + sub_chann_num = node.redis_connection.pubsub_numsub(channel) + if sub_chann_num == [(b_channel, 0)]: + sleep(0.3) + sub_chann_num = node.redis_connection.pubsub_numsub(channel) + assert sub_chann_num == [(b_channel, 1)] + # Assert that the cluster's pubsub_numsub function returns ALL clients + # subscribed to this channel in the entire cluster + assert r.pubsub_numsub(channel, target_nodes='all') == \ + [(b_channel, len(nodes))] + + def test_pubsub_numpat_merge_results(self, r): + nodes = r.get_nodes() + pubsub_nodes = [] + pattern = "foo*" + for node in nodes: + # We will create different pubsub clients where each one is + # connected to a different node + p = r.pubsub(node) + pubsub_nodes.append(p) + p.psubscribe(pattern) + # Assert that each node returns that only one client is subscribed + sub_num_pat = node.redis_connection.pubsub_numpat() + if sub_num_pat == 0: + sleep(0.3) + sub_num_pat = node.redis_connection.pubsub_numpat() + assert sub_num_pat == 1 + # Assert that the cluster's pubsub_numsub function returns ALL clients + # subscribed to this channel in the entire cluster + assert r.pubsub_numpat(target_nodes='all') == len(nodes) + + @skip_if_server_version_lt('2.8.0') + def test_cluster_pubsub_channels(self, r): + p = r.pubsub() + p.subscribe('foo', 'bar', 'baz', 'quux') + for i in range(4): + assert wait_for_message(p, timeout=0.5)['type'] == 'subscribe' + expected = [b'bar', b'baz', b'foo', b'quux'] + assert all([channel in r.pubsub_channels(target_nodes='all') + for channel in expected]) + + @skip_if_server_version_lt('2.8.0') + def test_cluster_pubsub_numsub(self, r): + p1 = r.pubsub() + p1.subscribe('foo', 'bar', 'baz') + for i in range(3): + assert wait_for_message(p1, timeout=0.5)['type'] == 'subscribe' + p2 = r.pubsub() + p2.subscribe('bar', 'baz') + for i in range(2): + assert wait_for_message(p2, timeout=0.5)['type'] == 'subscribe' + p3 = r.pubsub() + p3.subscribe('baz') + assert wait_for_message(p3, timeout=0.5)['type'] == 'subscribe' + + channels = [(b'foo', 1), (b'bar', 2), (b'baz', 3)] + assert r.pubsub_numsub('foo', 'bar', 'baz', target_nodes='all') \ + == channels + + def test_cluster_slots(self, r): + mock_all_nodes_resp(r, default_cluster_slots) + cluster_slots = r.cluster_slots() + assert isinstance(cluster_slots, dict) + assert len(default_cluster_slots) == len(cluster_slots) + assert cluster_slots.get((0, 8191)) is not None + assert cluster_slots.get((0, 8191)).get('primary') == \ + ('127.0.0.1', 7000) + + def test_cluster_addslots(self, r): + node = r.get_random_node() + mock_node_resp(node, 'OK') + assert r.cluster_addslots(node, 1, 2, 3) is True + + def test_cluster_countkeysinslot(self, r): + node = r.nodes_manager.get_node_from_slot(1) + mock_node_resp(node, 2) + assert r.cluster_countkeysinslot(1) == 2 + + def test_cluster_count_failure_report(self, r): + mock_all_nodes_resp(r, 0) + assert r.cluster_count_failure_report('node_0') == 0 + + def test_cluster_delslots(self): + cluster_slots = [ + [ + 0, 8191, + ['127.0.0.1', 7000, 'node_0'], + ], + [ + 8192, 16383, + ['127.0.0.1', 7001, 'node_1'], + ] + ] + r = get_mocked_redis_client(host=default_host, port=default_port, + cluster_slots=cluster_slots) + mock_all_nodes_resp(r, 'OK') + node0 = r.get_node(default_host, 7000) + node1 = r.get_node(default_host, 7001) + assert r.cluster_delslots(0, 8192) == [True, True] + assert node0.redis_connection.connection.read_response.called + assert node1.redis_connection.connection.read_response.called + + def test_cluster_failover(self, r): + node = r.get_random_node() + mock_node_resp(node, 'OK') + assert r.cluster_failover(node) is True + assert r.cluster_failover(node, 'FORCE') is True + assert r.cluster_failover(node, 'TAKEOVER') is True + with pytest.raises(RedisError): + r.cluster_failover(node, 'FORCT') + + def test_cluster_info(self, r): + info = r.cluster_info() + assert isinstance(info, dict) + assert info['cluster_state'] == 'ok' + + def test_cluster_keyslot(self, r): + mock_all_nodes_resp(r, 12182) + assert r.cluster_keyslot('foo') == 12182 + + def test_cluster_meet(self, r): + node = r.get_default_node() + mock_node_resp(node, 'OK') + assert r.cluster_meet('127.0.0.1', 6379) is True + + def test_cluster_nodes(self, r): + response = ( + 'c8253bae761cb1ecb2b61857d85dfe455a0fec8b 172.17.0.7:7006 ' + 'slave aa90da731f673a99617dfe930306549a09f83a6b 0 ' + '1447836263059 5 connected\n' + '9bd595fe4821a0e8d6b99d70faa660638a7612b3 172.17.0.7:7008 ' + 'master - 0 1447836264065 0 connected\n' + 'aa90da731f673a99617dfe930306549a09f83a6b 172.17.0.7:7003 ' + 'myself,master - 0 0 2 connected 5461-10922\n' + '1df047e5a594f945d82fc140be97a1452bcbf93e 172.17.0.7:7007 ' + 'slave 19efe5a631f3296fdf21a5441680f893e8cc96ec 0 ' + '1447836262556 3 connected\n' + '4ad9a12e63e8f0207025eeba2354bcf4c85e5b22 172.17.0.7:7005 ' + 'master - 0 1447836262555 7 connected 0-5460\n' + '19efe5a631f3296fdf21a5441680f893e8cc96ec 172.17.0.7:7004 ' + 'master - 0 1447836263562 3 connected 10923-16383\n' + 'fbb23ed8cfa23f17eaf27ff7d0c410492a1093d6 172.17.0.7:7002 ' + 'master,fail - 1447829446956 1447829444948 1 disconnected\n' + ) + mock_all_nodes_resp(r, response) + nodes = r.cluster_nodes() + assert len(nodes) == 7 + assert nodes.get('172.17.0.7:7006') is not None + assert nodes.get('172.17.0.7:7006').get('node_id') == \ + "c8253bae761cb1ecb2b61857d85dfe455a0fec8b" + + def test_cluster_replicate(self, r): + node = r.get_random_node() + all_replicas = r.get_replicas() + mock_all_nodes_resp(r, 'OK') + assert r.cluster_replicate(node, 'c8253bae761cb61857d') is True + results = r.cluster_replicate(all_replicas, 'c8253bae761cb61857d') + if isinstance(results, dict): + for res in results.values(): + assert res is True + else: + assert results is True + + def test_cluster_reset(self, r): + mock_all_nodes_resp(r, 'OK') + assert r.cluster_reset() is True + assert r.cluster_reset(False) is True + all_results = r.cluster_reset(False, target_nodes='all') + for res in all_results.values(): + assert res is True + + def test_cluster_save_config(self, r): + node = r.get_random_node() + all_nodes = r.get_nodes() + mock_all_nodes_resp(r, 'OK') + assert r.cluster_save_config(node) is True + all_results = r.cluster_save_config(all_nodes) + for res in all_results.values(): + assert res is True + + def test_cluster_get_keys_in_slot(self, r): + response = [b'{foo}1', b'{foo}2'] + node = r.nodes_manager.get_node_from_slot(12182) + mock_node_resp(node, response) + keys = r.cluster_get_keys_in_slot(12182, 4) + assert keys == response + + def test_cluster_set_config_epoch(self, r): + mock_all_nodes_resp(r, 'OK') + assert r.cluster_set_config_epoch(3) is True + all_results = r.cluster_set_config_epoch(3, target_nodes='all') + for res in all_results.values(): + assert res is True + + def test_cluster_setslot(self, r): + node = r.get_random_node() + mock_node_resp(node, 'OK') + assert r.cluster_setslot(node, 'node_0', 1218, 'IMPORTING') is True + assert r.cluster_setslot(node, 'node_0', 1218, 'NODE') is True + assert r.cluster_setslot(node, 'node_0', 1218, 'MIGRATING') is True + with pytest.raises(RedisError): + r.cluster_failover(node, 'STABLE') + with pytest.raises(RedisError): + r.cluster_failover(node, 'STATE') + + def test_cluster_setslot_stable(self, r): + node = r.nodes_manager.get_node_from_slot(12182) + mock_node_resp(node, 'OK') + assert r.cluster_setslot_stable(12182) is True + assert node.redis_connection.connection.read_response.called + + def test_cluster_replicas(self, r): + response = [b'01eca22229cf3c652b6fca0d09ff6941e0d2e3 ' + b'127.0.0.1:6377@16377 slave ' + b'52611e796814b78e90ad94be9d769a4f668f9a 0 ' + b'1634550063436 4 connected', + b'r4xfga22229cf3c652b6fca0d09ff69f3e0d4d ' + b'127.0.0.1:6378@16378 slave ' + b'52611e796814b78e90ad94be9d769a4f668f9a 0 ' + b'1634550063436 4 connected'] + mock_all_nodes_resp(r, response) + replicas = r.cluster_replicas('52611e796814b78e90ad94be9d769a4f668f9a') + assert replicas.get('127.0.0.1:6377') is not None + assert replicas.get('127.0.0.1:6378') is not None + assert replicas.get('127.0.0.1:6378').get('node_id') == \ + 'r4xfga22229cf3c652b6fca0d09ff69f3e0d4d' + + def test_readonly(self): + r = get_mocked_redis_client(host=default_host, port=default_port) + mock_all_nodes_resp(r, 'OK') + assert r.readonly() is True + all_replicas_results = r.readonly(target_nodes='replicas') + for res in all_replicas_results.values(): + assert res is True + for replica in r.get_replicas(): + assert replica.redis_connection.connection.read_response.called + + def test_readwrite(self): + r = get_mocked_redis_client(host=default_host, port=default_port) + mock_all_nodes_resp(r, 'OK') + assert r.readwrite() is True + all_replicas_results = r.readwrite(target_nodes='replicas') + for res in all_replicas_results.values(): + assert res is True + for replica in r.get_replicas(): + assert replica.redis_connection.connection.read_response.called + + def test_bgsave(self, r): + assert r.bgsave() + sleep(0.3) + assert r.bgsave(True) + + def test_info(self, r): + # Map keys to same slot + r.set('x{1}', 1) + r.set('y{1}', 2) + r.set('z{1}', 3) + # Get node that handles the slot + slot = r.keyslot('x{1}') + node = r.nodes_manager.get_node_from_slot(slot) + # Run info on that node + info = r.info(target_nodes=node) + assert isinstance(info, dict) + assert info['db0']['keys'] == 3 + + def _init_slowlog_test(self, r, node): + slowlog_lim = r.config_get('slowlog-log-slower-than', + target_nodes=node) + assert r.config_set('slowlog-log-slower-than', 0, target_nodes=node) \ + is True + return slowlog_lim['slowlog-log-slower-than'] + + def _teardown_slowlog_test(self, r, node, prev_limit): + assert r.config_set('slowlog-log-slower-than', prev_limit, + target_nodes=node) is True + + def test_slowlog_get(self, r, slowlog): + unicode_string = chr(3456) + 'abcd' + chr(3421) + node = r.get_node_from_key(unicode_string) + slowlog_limit = self._init_slowlog_test(r, node) + assert r.slowlog_reset(target_nodes=node) + r.get(unicode_string) + slowlog = r.slowlog_get(target_nodes=node) + assert isinstance(slowlog, list) + commands = [log['command'] for log in slowlog] + + get_command = b' '.join((b'GET', unicode_string.encode('utf-8'))) + assert get_command in commands + assert b'SLOWLOG RESET' in commands + + # the order should be ['GET ', 'SLOWLOG RESET'], + # but if other clients are executing commands at the same time, there + # could be commands, before, between, or after, so just check that + # the two we care about are in the appropriate order. + assert commands.index(get_command) < commands.index(b'SLOWLOG RESET') + + # make sure other attributes are typed correctly + assert isinstance(slowlog[0]['start_time'], int) + assert isinstance(slowlog[0]['duration'], int) + # rollback the slowlog limit to its original value + self._teardown_slowlog_test(r, node, slowlog_limit) + + def test_slowlog_get_limit(self, r, slowlog): + assert r.slowlog_reset() + node = r.get_node_from_key('foo') + slowlog_limit = self._init_slowlog_test(r, node) + r.get('foo') + slowlog = r.slowlog_get(1, target_nodes=node) + assert isinstance(slowlog, list) + # only one command, based on the number we passed to slowlog_get() + assert len(slowlog) == 1 + self._teardown_slowlog_test(r, node, slowlog_limit) + + def test_slowlog_length(self, r, slowlog): + r.get('foo') + node = r.nodes_manager.get_node_from_slot(key_slot(b'foo')) + slowlog_len = r.slowlog_len(target_nodes=node) + assert isinstance(slowlog_len, int) + + def test_time(self, r): + t = r.time(target_nodes=r.get_primaries()[0]) + assert len(t) == 2 + assert isinstance(t[0], int) + assert isinstance(t[1], int) + + @skip_if_server_version_lt('4.0.0') + def test_memory_usage(self, r): + r.set('foo', 'bar') + assert isinstance(r.memory_usage('foo'), int) + + @skip_if_server_version_lt('4.0.0') + def test_memory_malloc_stats(self, r): + assert r.memory_malloc_stats() + + @skip_if_server_version_lt('4.0.0') + def test_memory_stats(self, r): + # put a key into the current db to make sure that "db." + # has data + r.set('foo', 'bar') + node = r.nodes_manager.get_node_from_slot(key_slot(b'foo')) + stats = r.memory_stats(target_nodes=node) + assert isinstance(stats, dict) + for key, value in stats.items(): + if key.startswith('db.'): + assert isinstance(value, dict) + + @skip_if_server_version_lt('4.0.0') + def test_memory_help(self, r): + with pytest.raises(NotImplementedError): + r.memory_help() + + @skip_if_server_version_lt('4.0.0') + def test_memory_doctor(self, r): + with pytest.raises(NotImplementedError): + r.memory_doctor() + + def test_lastsave(self, r): + node = r.get_primaries()[0] + assert isinstance(r.lastsave(target_nodes=node), + datetime.datetime) + + def test_cluster_echo(self, r): + node = r.get_primaries()[0] + assert r.echo('foo bar', node) == b'foo bar' + + @skip_if_server_version_lt('1.0.0') + def test_debug_segfault(self, r): + with pytest.raises(NotImplementedError): + r.debug_segfault() + + def test_config_resetstat(self, r): + node = r.get_primaries()[0] + r.ping(target_nodes=node) + prior_commands_processed = \ + int(r.info(target_nodes=node)['total_commands_processed']) + assert prior_commands_processed >= 1 + r.config_resetstat(target_nodes=node) + reset_commands_processed = \ + int(r.info(target_nodes=node)['total_commands_processed']) + assert reset_commands_processed < prior_commands_processed + + @skip_if_server_version_lt('6.2.0') + def test_client_trackinginfo(self, r): + node = r.get_primaries()[0] + res = r.client_trackinginfo(target_nodes=node) + assert len(res) > 2 + assert 'prefixes' in res + + @skip_if_server_version_lt('2.9.50') + def test_client_pause(self, r): + node = r.get_primaries()[0] + assert r.client_pause(1, target_nodes=node) + assert r.client_pause(timeout=1, target_nodes=node) + with pytest.raises(RedisError): + r.client_pause(timeout='not an integer', target_nodes=node) + + @skip_if_server_version_lt('6.2.0') + def test_client_unpause(self, r): + assert r.client_unpause() + + @skip_if_server_version_lt('5.0.0') + def test_client_id(self, r): + node = r.get_primaries()[0] + assert r.client_id(target_nodes=node) > 0 + + @skip_if_server_version_lt('5.0.0') + def test_client_unblock(self, r): + node = r.get_primaries()[0] + myid = r.client_id(target_nodes=node) + assert not r.client_unblock(myid, target_nodes=node) + assert not r.client_unblock(myid, error=True, target_nodes=node) + assert not r.client_unblock(myid, error=False, target_nodes=node) + + @skip_if_server_version_lt('6.0.0') + def test_client_getredir(self, r): + node = r.get_primaries()[0] + assert isinstance(r.client_getredir(target_nodes=node), int) + assert r.client_getredir(target_nodes=node) == -1 + + @skip_if_server_version_lt('6.2.0') + def test_client_info(self, r): + node = r.get_primaries()[0] + info = r.client_info(target_nodes=node) + assert isinstance(info, dict) + assert 'addr' in info + + @skip_if_server_version_lt('2.6.9') + def test_client_kill(self, r, r2): + node = r.get_primaries()[0] + r.client_setname('redis-py-c1', target_nodes='all') + r2.client_setname('redis-py-c2', target_nodes='all') + clients = [client for client in r.client_list(target_nodes=node) + if client.get('name') in ['redis-py-c1', 'redis-py-c2']] + assert len(clients) == 2 + clients_by_name = dict([(client.get('name'), client) + for client in clients]) + + client_addr = clients_by_name['redis-py-c2'].get('addr') + assert r.client_kill(client_addr, target_nodes=node) is True + + clients = [client for client in r.client_list(target_nodes=node) + if client.get('name') in ['redis-py-c1', 'redis-py-c2']] + assert len(clients) == 1 + assert clients[0].get('name') == 'redis-py-c1' + + @skip_if_server_version_lt('2.6.0') + def test_cluster_bitop_not_empty_string(self, r): + r['{foo}a'] = '' + r.bitop('not', '{foo}r', '{foo}a') + assert r.get('{foo}r') is None + + @skip_if_server_version_lt('2.6.0') + def test_cluster_bitop_not(self, r): + test_str = b'\xAA\x00\xFF\x55' + correct = ~0xAA00FF55 & 0xFFFFFFFF + r['{foo}a'] = test_str + r.bitop('not', '{foo}r', '{foo}a') + assert int(binascii.hexlify(r['{foo}r']), 16) == correct + + @skip_if_server_version_lt('2.6.0') + def test_cluster_bitop_not_in_place(self, r): + test_str = b'\xAA\x00\xFF\x55' + correct = ~0xAA00FF55 & 0xFFFFFFFF + r['{foo}a'] = test_str + r.bitop('not', '{foo}a', '{foo}a') + assert int(binascii.hexlify(r['{foo}a']), 16) == correct + + @skip_if_server_version_lt('2.6.0') + def test_cluster_bitop_single_string(self, r): + test_str = b'\x01\x02\xFF' + r['{foo}a'] = test_str + r.bitop('and', '{foo}res1', '{foo}a') + r.bitop('or', '{foo}res2', '{foo}a') + r.bitop('xor', '{foo}res3', '{foo}a') + assert r['{foo}res1'] == test_str + assert r['{foo}res2'] == test_str + assert r['{foo}res3'] == test_str + + @skip_if_server_version_lt('2.6.0') + def test_cluster_bitop_string_operands(self, r): + r['{foo}a'] = b'\x01\x02\xFF\xFF' + r['{foo}b'] = b'\x01\x02\xFF' + r.bitop('and', '{foo}res1', '{foo}a', '{foo}b') + r.bitop('or', '{foo}res2', '{foo}a', '{foo}b') + r.bitop('xor', '{foo}res3', '{foo}a', '{foo}b') + assert int(binascii.hexlify(r['{foo}res1']), 16) == 0x0102FF00 + assert int(binascii.hexlify(r['{foo}res2']), 16) == 0x0102FFFF + assert int(binascii.hexlify(r['{foo}res3']), 16) == 0x000000FF + + @skip_if_server_version_lt('6.2.0') + def test_cluster_copy(self, r): + assert r.copy("{foo}a", "{foo}b") == 0 + r.set("{foo}a", "bar") + assert r.copy("{foo}a", "{foo}b") == 1 + assert r.get("{foo}a") == b"bar" + assert r.get("{foo}b") == b"bar" + + @skip_if_server_version_lt('6.2.0') + def test_cluster_copy_and_replace(self, r): + r.set("{foo}a", "foo1") + r.set("{foo}b", "foo2") + assert r.copy("{foo}a", "{foo}b") == 0 + assert r.copy("{foo}a", "{foo}b", replace=True) == 1 + + @skip_if_server_version_lt('6.2.0') + def test_cluster_lmove(self, r): + r.rpush('{foo}a', 'one', 'two', 'three', 'four') + assert r.lmove('{foo}a', '{foo}b') + assert r.lmove('{foo}a', '{foo}b', 'right', 'left') + + @skip_if_server_version_lt('6.2.0') + def test_cluster_blmove(self, r): + r.rpush('{foo}a', 'one', 'two', 'three', 'four') + assert r.blmove('{foo}a', '{foo}b', 5) + assert r.blmove('{foo}a', '{foo}b', 1, 'RIGHT', 'LEFT') + + def test_cluster_msetnx(self, r): + d = {'{foo}a': b'1', '{foo}b': b'2', '{foo}c': b'3'} + assert r.msetnx(d) + d2 = {'{foo}a': b'x', '{foo}d': b'4'} + assert not r.msetnx(d2) + for k, v in d.items(): + assert r[k] == v + assert r.get('{foo}d') is None + + def test_cluster_rename(self, r): + r['{foo}a'] = '1' + assert r.rename('{foo}a', '{foo}b') + assert r.get('{foo}a') is None + assert r['{foo}b'] == b'1' + + def test_cluster_renamenx(self, r): + r['{foo}a'] = '1' + r['{foo}b'] = '2' + assert not r.renamenx('{foo}a', '{foo}b') + assert r['{foo}a'] == b'1' + assert r['{foo}b'] == b'2' + + # LIST COMMANDS + def test_cluster_blpop(self, r): + r.rpush('{foo}a', '1', '2') + r.rpush('{foo}b', '3', '4') + assert r.blpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}b', b'3') + assert r.blpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}b', b'4') + assert r.blpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}a', b'1') + assert r.blpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}a', b'2') + assert r.blpop(['{foo}b', '{foo}a'], timeout=1) is None + r.rpush('{foo}c', '1') + assert r.blpop('{foo}c', timeout=1) == (b'{foo}c', b'1') + + def test_cluster_brpop(self, r): + r.rpush('{foo}a', '1', '2') + r.rpush('{foo}b', '3', '4') + assert r.brpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}b', b'4') + assert r.brpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}b', b'3') + assert r.brpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}a', b'2') + assert r.brpop(['{foo}b', '{foo}a'], timeout=1) == (b'{foo}a', b'1') + assert r.brpop(['{foo}b', '{foo}a'], timeout=1) is None + r.rpush('{foo}c', '1') + assert r.brpop('{foo}c', timeout=1) == (b'{foo}c', b'1') + + def test_cluster_brpoplpush(self, r): + r.rpush('{foo}a', '1', '2') + r.rpush('{foo}b', '3', '4') + assert r.brpoplpush('{foo}a', '{foo}b') == b'2' + assert r.brpoplpush('{foo}a', '{foo}b') == b'1' + assert r.brpoplpush('{foo}a', '{foo}b', timeout=1) is None + assert r.lrange('{foo}a', 0, -1) == [] + assert r.lrange('{foo}b', 0, -1) == [b'1', b'2', b'3', b'4'] + + def test_cluster_brpoplpush_empty_string(self, r): + r.rpush('{foo}a', '') + assert r.brpoplpush('{foo}a', '{foo}b') == b'' + + def test_cluster_rpoplpush(self, r): + r.rpush('{foo}a', 'a1', 'a2', 'a3') + r.rpush('{foo}b', 'b1', 'b2', 'b3') + assert r.rpoplpush('{foo}a', '{foo}b') == b'a3' + assert r.lrange('{foo}a', 0, -1) == [b'a1', b'a2'] + assert r.lrange('{foo}b', 0, -1) == [b'a3', b'b1', b'b2', b'b3'] + + def test_cluster_sdiff(self, r): + r.sadd('{foo}a', '1', '2', '3') + assert r.sdiff('{foo}a', '{foo}b') == {b'1', b'2', b'3'} + r.sadd('{foo}b', '2', '3') + assert r.sdiff('{foo}a', '{foo}b') == {b'1'} + + def test_cluster_sdiffstore(self, r): + r.sadd('{foo}a', '1', '2', '3') + assert r.sdiffstore('{foo}c', '{foo}a', '{foo}b') == 3 + assert r.smembers('{foo}c') == {b'1', b'2', b'3'} + r.sadd('{foo}b', '2', '3') + assert r.sdiffstore('{foo}c', '{foo}a', '{foo}b') == 1 + assert r.smembers('{foo}c') == {b'1'} + + def test_cluster_sinter(self, r): + r.sadd('{foo}a', '1', '2', '3') + assert r.sinter('{foo}a', '{foo}b') == set() + r.sadd('{foo}b', '2', '3') + assert r.sinter('{foo}a', '{foo}b') == {b'2', b'3'} + + def test_cluster_sinterstore(self, r): + r.sadd('{foo}a', '1', '2', '3') + assert r.sinterstore('{foo}c', '{foo}a', '{foo}b') == 0 + assert r.smembers('{foo}c') == set() + r.sadd('{foo}b', '2', '3') + assert r.sinterstore('{foo}c', '{foo}a', '{foo}b') == 2 + assert r.smembers('{foo}c') == {b'2', b'3'} + + def test_cluster_smove(self, r): + r.sadd('{foo}a', 'a1', 'a2') + r.sadd('{foo}b', 'b1', 'b2') + assert r.smove('{foo}a', '{foo}b', 'a1') + assert r.smembers('{foo}a') == {b'a2'} + assert r.smembers('{foo}b') == {b'b1', b'b2', b'a1'} + + def test_cluster_sunion(self, r): + r.sadd('{foo}a', '1', '2') + r.sadd('{foo}b', '2', '3') + assert r.sunion('{foo}a', '{foo}b') == {b'1', b'2', b'3'} + + def test_cluster_sunionstore(self, r): + r.sadd('{foo}a', '1', '2') + r.sadd('{foo}b', '2', '3') + assert r.sunionstore('{foo}c', '{foo}a', '{foo}b') == 3 + assert r.smembers('{foo}c') == {b'1', b'2', b'3'} + + @skip_if_server_version_lt('6.2.0') + def test_cluster_zdiff(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2, 'a3': 3}) + r.zadd('{foo}b', {'a1': 1, 'a2': 2}) + assert r.zdiff(['{foo}a', '{foo}b']) == [b'a3'] + assert r.zdiff(['{foo}a', '{foo}b'], withscores=True) == [b'a3', b'3'] + + @skip_if_server_version_lt('6.2.0') + def test_cluster_zdiffstore(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2, 'a3': 3}) + r.zadd('{foo}b', {'a1': 1, 'a2': 2}) + assert r.zdiffstore("{foo}out", ['{foo}a', '{foo}b']) + assert r.zrange("{foo}out", 0, -1) == [b'a3'] + assert r.zrange("{foo}out", 0, -1, withscores=True) == [(b'a3', 3.0)] + + @skip_if_server_version_lt('6.2.0') + def test_cluster_zinter(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zinter(['{foo}a', '{foo}b', '{foo}c']) == [b'a3', b'a1'] + # invalid aggregation + with pytest.raises(DataError): + r.zinter(['{foo}a', '{foo}b', '{foo}c'], + aggregate='foo', withscores=True) + # aggregate with SUM + assert r.zinter(['{foo}a', '{foo}b', '{foo}c'], withscores=True) \ + == [(b'a3', 8), (b'a1', 9)] + # aggregate with MAX + assert r.zinter(['{foo}a', '{foo}b', '{foo}c'], aggregate='MAX', + withscores=True) \ + == [(b'a3', 5), (b'a1', 6)] + # aggregate with MIN + assert r.zinter(['{foo}a', '{foo}b', '{foo}c'], aggregate='MIN', + withscores=True) \ + == [(b'a1', 1), (b'a3', 1)] + # with weights + assert r.zinter({'{foo}a': 1, '{foo}b': 2, '{foo}c': 3}, + withscores=True) \ + == [(b'a3', 20), (b'a1', 23)] + + def test_cluster_zinterstore_sum(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zinterstore('{foo}d', ['{foo}a', '{foo}b', '{foo}c']) == 2 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a3', 8), (b'a1', 9)] + + def test_cluster_zinterstore_max(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zinterstore( + '{foo}d', ['{foo}a', '{foo}b', '{foo}c'], aggregate='MAX') == 2 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a3', 5), (b'a1', 6)] + + def test_cluster_zinterstore_min(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2, 'a3': 3}) + r.zadd('{foo}b', {'a1': 2, 'a2': 3, 'a3': 5}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zinterstore( + '{foo}d', ['{foo}a', '{foo}b', '{foo}c'], aggregate='MIN') == 2 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a1', 1), (b'a3', 3)] + + def test_cluster_zinterstore_with_weight(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zinterstore( + '{foo}d', {'{foo}a': 1, '{foo}b': 2, '{foo}c': 3}) == 2 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a3', 20), (b'a1', 23)] + + @skip_if_server_version_lt('4.9.0') + def test_cluster_bzpopmax(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2}) + r.zadd('{foo}b', {'b1': 10, 'b2': 20}) + assert r.bzpopmax(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}b', b'b2', 20) + assert r.bzpopmax(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}b', b'b1', 10) + assert r.bzpopmax(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}a', b'a2', 2) + assert r.bzpopmax(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}a', b'a1', 1) + assert r.bzpopmax(['{foo}b', '{foo}a'], timeout=1) is None + r.zadd('{foo}c', {'c1': 100}) + assert r.bzpopmax('{foo}c', timeout=1) == (b'{foo}c', b'c1', 100) + + @skip_if_server_version_lt('4.9.0') + def test_cluster_bzpopmin(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2}) + r.zadd('{foo}b', {'b1': 10, 'b2': 20}) + assert r.bzpopmin(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}b', b'b1', 10) + assert r.bzpopmin(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}b', b'b2', 20) + assert r.bzpopmin(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}a', b'a1', 1) + assert r.bzpopmin(['{foo}b', '{foo}a'], timeout=1) == ( + b'{foo}a', b'a2', 2) + assert r.bzpopmin(['{foo}b', '{foo}a'], timeout=1) is None + r.zadd('{foo}c', {'c1': 100}) + assert r.bzpopmin('{foo}c', timeout=1) == (b'{foo}c', b'c1', 100) + + @skip_if_server_version_lt('6.2.0') + def test_cluster_zrangestore(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2, 'a3': 3}) + assert r.zrangestore('{foo}b', '{foo}a', 0, 1) + assert r.zrange('{foo}b', 0, -1) == [b'a1', b'a2'] + assert r.zrangestore('{foo}b', '{foo}a', 1, 2) + assert r.zrange('{foo}b', 0, -1) == [b'a2', b'a3'] + assert r.zrange('{foo}b', 0, -1, withscores=True) == \ + [(b'a2', 2), (b'a3', 3)] + # reversed order + assert r.zrangestore('{foo}b', '{foo}a', 1, 2, desc=True) + assert r.zrange('{foo}b', 0, -1) == [b'a1', b'a2'] + # by score + assert r.zrangestore('{foo}b', '{foo}a', 2, 1, byscore=True, + offset=0, num=1, desc=True) + assert r.zrange('{foo}b', 0, -1) == [b'a2'] + # by lex + assert r.zrangestore('{foo}b', '{foo}a', '[a2', '(a3', bylex=True, + offset=0, num=1) + assert r.zrange('{foo}b', 0, -1) == [b'a2'] + + @skip_if_server_version_lt('6.2.0') + def test_cluster_zunion(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + # sum + assert r.zunion(['{foo}a', '{foo}b', '{foo}c']) == \ + [b'a2', b'a4', b'a3', b'a1'] + assert r.zunion(['{foo}a', '{foo}b', '{foo}c'], withscores=True) == \ + [(b'a2', 3), (b'a4', 4), (b'a3', 8), (b'a1', 9)] + # max + assert r.zunion(['{foo}a', '{foo}b', '{foo}c'], aggregate='MAX', + withscores=True) \ + == [(b'a2', 2), (b'a4', 4), (b'a3', 5), (b'a1', 6)] + # min + assert r.zunion(['{foo}a', '{foo}b', '{foo}c'], aggregate='MIN', + withscores=True) \ + == [(b'a1', 1), (b'a2', 1), (b'a3', 1), (b'a4', 4)] + # with weight + assert r.zunion({'{foo}a': 1, '{foo}b': 2, '{foo}c': 3}, + withscores=True) \ + == [(b'a2', 5), (b'a4', 12), (b'a3', 20), (b'a1', 23)] + + def test_cluster_zunionstore_sum(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zunionstore('{foo}d', ['{foo}a', '{foo}b', '{foo}c']) == 4 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a2', 3), (b'a4', 4), (b'a3', 8), (b'a1', 9)] + + def test_cluster_zunionstore_max(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zunionstore( + '{foo}d', ['{foo}a', '{foo}b', '{foo}c'], aggregate='MAX') == 4 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a2', 2), (b'a4', 4), (b'a3', 5), (b'a1', 6)] + + def test_cluster_zunionstore_min(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 2, 'a3': 3}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 4}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zunionstore( + '{foo}d', ['{foo}a', '{foo}b', '{foo}c'], aggregate='MIN') == 4 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a1', 1), (b'a2', 2), (b'a3', 3), (b'a4', 4)] + + def test_cluster_zunionstore_with_weight(self, r): + r.zadd('{foo}a', {'a1': 1, 'a2': 1, 'a3': 1}) + r.zadd('{foo}b', {'a1': 2, 'a2': 2, 'a3': 2}) + r.zadd('{foo}c', {'a1': 6, 'a3': 5, 'a4': 4}) + assert r.zunionstore( + '{foo}d', {'{foo}a': 1, '{foo}b': 2, '{foo}c': 3}) == 4 + assert r.zrange('{foo}d', 0, -1, withscores=True) == \ + [(b'a2', 5), (b'a4', 12), (b'a3', 20), (b'a1', 23)] + + @skip_if_server_version_lt('2.8.9') + def test_cluster_pfcount(self, r): + members = {b'1', b'2', b'3'} + r.pfadd('{foo}a', *members) + assert r.pfcount('{foo}a') == len(members) + members_b = {b'2', b'3', b'4'} + r.pfadd('{foo}b', *members_b) + assert r.pfcount('{foo}b') == len(members_b) + assert r.pfcount('{foo}a', '{foo}b') == len(members_b.union(members)) + + @skip_if_server_version_lt('2.8.9') + def test_cluster_pfmerge(self, r): + mema = {b'1', b'2', b'3'} + memb = {b'2', b'3', b'4'} + memc = {b'5', b'6', b'7'} + r.pfadd('{foo}a', *mema) + r.pfadd('{foo}b', *memb) + r.pfadd('{foo}c', *memc) + r.pfmerge('{foo}d', '{foo}c', '{foo}a') + assert r.pfcount('{foo}d') == 6 + r.pfmerge('{foo}d', '{foo}b') + assert r.pfcount('{foo}d') == 7 + + def test_cluster_sort_store(self, r): + r.rpush('{foo}a', '2', '3', '1') + assert r.sort('{foo}a', store='{foo}sorted_values') == 3 + assert r.lrange('{foo}sorted_values', 0, -1) == [b'1', b'2', b'3'] + + # GEO COMMANDS + @skip_if_server_version_lt('6.2.0') + def test_cluster_geosearchstore(self, r): + values = (2.1909389952632, 41.433791470673, 'place1') + \ + (2.1873744593677, 41.406342043777, 'place2') + + r.geoadd('{foo}barcelona', values) + r.geosearchstore('{foo}places_barcelona', '{foo}barcelona', + longitude=2.191, latitude=41.433, radius=1000) + assert r.zrange('{foo}places_barcelona', 0, -1) == [b'place1'] + + @skip_unless_arch_bits(64) + @skip_if_server_version_lt('6.2.0') + def test_geosearchstore_dist(self, r): + values = (2.1909389952632, 41.433791470673, 'place1') + \ + (2.1873744593677, 41.406342043777, 'place2') + + r.geoadd('{foo}barcelona', values) + r.geosearchstore('{foo}places_barcelona', '{foo}barcelona', + longitude=2.191, latitude=41.433, + radius=1000, storedist=True) + # instead of save the geo score, the distance is saved. + assert r.zscore('{foo}places_barcelona', 'place1') == 88.05060698409301 + + @skip_if_server_version_lt('3.2.0') + def test_cluster_georadius_store(self, r): + values = (2.1909389952632, 41.433791470673, 'place1') + \ + (2.1873744593677, 41.406342043777, 'place2') + + r.geoadd('{foo}barcelona', values) + r.georadius('{foo}barcelona', 2.191, 41.433, + 1000, store='{foo}places_barcelona') + assert r.zrange('{foo}places_barcelona', 0, -1) == [b'place1'] + + @skip_unless_arch_bits(64) + @skip_if_server_version_lt('3.2.0') + def test_cluster_georadius_store_dist(self, r): + values = (2.1909389952632, 41.433791470673, 'place1') + \ + (2.1873744593677, 41.406342043777, 'place2') + + r.geoadd('{foo}barcelona', values) + r.georadius('{foo}barcelona', 2.191, 41.433, 1000, + store_dist='{foo}places_barcelona') + # instead of save the geo score, the distance is saved. + assert r.zscore('{foo}places_barcelona', 'place1') == 88.05060698409301 + + def test_cluster_dbsize(self, r): + d = {'a': b'1', 'b': b'2', 'c': b'3', 'd': b'4'} + assert r.mset_nonatomic(d) + assert r.dbsize(target_nodes='primaries') == len(d) + + def test_cluster_keys(self, r): + assert r.keys() == [] + keys_with_underscores = {b'test_a', b'test_b'} + keys = keys_with_underscores.union({b'testc'}) + for key in keys: + r[key] = 1 + assert set(r.keys(pattern='test_*', target_nodes='primaries')) == \ + keys_with_underscores + assert set(r.keys(pattern='test*', target_nodes='primaries')) == keys + + # SCAN COMMANDS + @skip_if_server_version_lt('2.8.0') + def test_cluster_scan(self, r): + r.set('a', 1) + r.set('b', 2) + r.set('c', 3) + cursor, keys = r.scan(target_nodes='primaries') + assert cursor == 0 + assert set(keys) == {b'a', b'b', b'c'} + _, keys = r.scan(match='a', target_nodes='primaries') + assert set(keys) == {b'a'} + + @skip_if_server_version_lt("6.0.0") + def test_cluster_scan_type(self, r): + r.sadd('a-set', 1) + r.hset('a-hash', 'foo', 2) + r.lpush('a-list', 'aux', 3) + _, keys = r.scan(match='a*', _type='SET', target_nodes='primaries') + assert set(keys) == {b'a-set'} + + @skip_if_server_version_lt('2.8.0') + def test_cluster_scan_iter(self, r): + r.set('a', 1) + r.set('b', 2) + r.set('c', 3) + keys = list(r.scan_iter(target_nodes='primaries')) + assert set(keys) == {b'a', b'b', b'c'} + keys = list(r.scan_iter(match='a', target_nodes='primaries')) + assert set(keys) == {b'a'} + + def test_cluster_randomkey(self, r): + node = r.get_node_from_key('{foo}') + assert r.randomkey(target_nodes=node) is None + for key in ('{foo}a', '{foo}b', '{foo}c'): + r[key] = 1 + assert r.randomkey(target_nodes=node) in \ + (b'{foo}a', b'{foo}b', b'{foo}c') + + +@pytest.mark.onlycluster +class TestNodesManager: + """ + Tests for the NodesManager class + """ + + def test_load_balancer(self, r): + n_manager = r.nodes_manager + lb = n_manager.read_load_balancer + slot_1 = 1257 + slot_2 = 8975 + node_1 = ClusterNode(default_host, 6379, PRIMARY) + node_2 = ClusterNode(default_host, 6378, REPLICA) + node_3 = ClusterNode(default_host, 6377, REPLICA) + node_4 = ClusterNode(default_host, 6376, PRIMARY) + node_5 = ClusterNode(default_host, 6375, REPLICA) + n_manager.slots_cache = { + slot_1: [node_1, node_2, node_3], + slot_2: [node_4, node_5] + } + primary1_name = n_manager.slots_cache[slot_1][0].name + primary2_name = n_manager.slots_cache[slot_2][0].name + list1_size = len(n_manager.slots_cache[slot_1]) + list2_size = len(n_manager.slots_cache[slot_2]) + # slot 1 + assert lb.get_server_index(primary1_name, list1_size) == 0 + assert lb.get_server_index(primary1_name, list1_size) == 1 + assert lb.get_server_index(primary1_name, list1_size) == 2 + assert lb.get_server_index(primary1_name, list1_size) == 0 + # slot 2 + assert lb.get_server_index(primary2_name, list2_size) == 0 + assert lb.get_server_index(primary2_name, list2_size) == 1 + assert lb.get_server_index(primary2_name, list2_size) == 0 + + lb.reset() + assert lb.get_server_index(primary1_name, list1_size) == 0 + assert lb.get_server_index(primary2_name, list2_size) == 0 + + def test_init_slots_cache_not_all_slots_covered(self): + """ + Test that if not all slots are covered it should raise an exception + """ + # Missing slot 5460 + cluster_slots = [ + [0, 5459, ['127.0.0.1', 7000], ['127.0.0.1', 7003]], + [5461, 10922, ['127.0.0.1', 7001], + ['127.0.0.1', 7004]], + [10923, 16383, ['127.0.0.1', 7002], + ['127.0.0.1', 7005]], + ] + with pytest.raises(RedisClusterException) as ex: + get_mocked_redis_client(host=default_host, port=default_port, + cluster_slots=cluster_slots) + assert str(ex.value).startswith( + "All slots are not covered after query all startup_nodes.") + + def test_init_slots_cache_not_require_full_coverage_error(self): + """ + When require_full_coverage is set to False and not all slots are + covered, if one of the nodes has 'cluster-require_full_coverage' + config set to 'yes' the cluster initialization should fail + """ + # Missing slot 5460 + cluster_slots = [ + [0, 5459, ['127.0.0.1', 7000], ['127.0.0.1', 7003]], + [5461, 10922, ['127.0.0.1', 7001], + ['127.0.0.1', 7004]], + [10923, 16383, ['127.0.0.1', 7002], + ['127.0.0.1', 7005]], + ] + + with pytest.raises(RedisClusterException): + get_mocked_redis_client(host=default_host, port=default_port, + cluster_slots=cluster_slots, + require_full_coverage=False, + coverage_result='yes') + + def test_init_slots_cache_not_require_full_coverage_success(self): + """ + When require_full_coverage is set to False and not all slots are + covered, if all of the nodes has 'cluster-require_full_coverage' + config set to 'no' the cluster initialization should succeed + """ + # Missing slot 5460 + cluster_slots = [ + [0, 5459, ['127.0.0.1', 7000], ['127.0.0.1', 7003]], + [5461, 10922, ['127.0.0.1', 7001], + ['127.0.0.1', 7004]], + [10923, 16383, ['127.0.0.1', 7002], + ['127.0.0.1', 7005]], + ] + + rc = get_mocked_redis_client(host=default_host, port=default_port, + cluster_slots=cluster_slots, + require_full_coverage=False, + coverage_result='no') + + assert 5460 not in rc.nodes_manager.slots_cache + + def test_init_slots_cache_not_require_full_coverage_skips_check(self): + """ + Test that when require_full_coverage is set to False and + skip_full_coverage_check is set to true, the cluster initialization + succeed without checking the nodes' Redis configurations + """ + # Missing slot 5460 + cluster_slots = [ + [0, 5459, ['127.0.0.1', 7000], ['127.0.0.1', 7003]], + [5461, 10922, ['127.0.0.1', 7001], + ['127.0.0.1', 7004]], + [10923, 16383, ['127.0.0.1', 7002], + ['127.0.0.1', 7005]], + ] + + with patch.object(NodesManager, + 'cluster_require_full_coverage') as conf_check_mock: + rc = get_mocked_redis_client(host=default_host, port=default_port, + cluster_slots=cluster_slots, + require_full_coverage=False, + skip_full_coverage_check=True, + coverage_result='no') + + assert conf_check_mock.called is False + assert 5460 not in rc.nodes_manager.slots_cache + + def test_init_slots_cache(self): + """ + Test that slots cache can in initialized and all slots are covered + """ + good_slots_resp = [ + [0, 5460, ['127.0.0.1', 7000], ['127.0.0.2', 7003]], + [5461, 10922, ['127.0.0.1', 7001], ['127.0.0.2', 7004]], + [10923, 16383, ['127.0.0.1', 7002], ['127.0.0.2', 7005]], + ] + + rc = get_mocked_redis_client(host=default_host, port=default_port, + cluster_slots=good_slots_resp) + n_manager = rc.nodes_manager + assert len(n_manager.slots_cache) == REDIS_CLUSTER_HASH_SLOTS + for slot_info in good_slots_resp: + all_hosts = ['127.0.0.1', '127.0.0.2'] + all_ports = [7000, 7001, 7002, 7003, 7004, 7005] + slot_start = slot_info[0] + slot_end = slot_info[1] + for i in range(slot_start, slot_end + 1): + assert len(n_manager.slots_cache[i]) == len(slot_info[2:]) + assert n_manager.slots_cache[i][0].host in all_hosts + assert n_manager.slots_cache[i][1].host in all_hosts + assert n_manager.slots_cache[i][0].port in all_ports + assert n_manager.slots_cache[i][1].port in all_ports + + assert len(n_manager.nodes_cache) == 6 + + def test_empty_startup_nodes(self): + """ + It should not be possible to create a node manager with no nodes + specified + """ + with pytest.raises(RedisClusterException): + NodesManager([]) + + def test_wrong_startup_nodes_type(self): + """ + If something other then a list type itteratable is provided it should + fail + """ + with pytest.raises(RedisClusterException): + NodesManager({}) + + def test_init_slots_cache_slots_collision(self, request): + """ + Test that if 2 nodes do not agree on the same slots setup it should + raise an error. In this test both nodes will say that the first + slots block should be bound to different servers. + """ + with patch.object(NodesManager, + 'create_redis_node') as create_redis_node: + def create_mocked_redis_node(host, port, **kwargs): + """ + Helper function to return custom slots cache data from + different redis nodes + """ + if port == 7000: + result = [ + [ + 0, + 5460, + ['127.0.0.1', 7000], + ['127.0.0.1', 7003], + ], + [ + 5461, + 10922, + ['127.0.0.1', 7001], + ['127.0.0.1', 7004], + ], + ] + + elif port == 7001: + result = [ + [ + 0, + 5460, + ['127.0.0.1', 7001], + ['127.0.0.1', 7003], + ], + [ + 5461, + 10922, + ['127.0.0.1', 7000], + ['127.0.0.1', 7004], + ], + ] + else: + result = [] + + r_node = Redis( + host=host, + port=port + ) + + orig_execute_command = r_node.execute_command + + def execute_command(*args, **kwargs): + if args[0] == 'CLUSTER SLOTS': + return result + elif args[1] == 'cluster-require-full-coverage': + return {'cluster-require-full-coverage': 'yes'} + else: + return orig_execute_command(*args, **kwargs) + + r_node.execute_command = execute_command + return r_node + + create_redis_node.side_effect = create_mocked_redis_node + + with pytest.raises(RedisClusterException) as ex: + node_1 = ClusterNode('127.0.0.1', 7000) + node_2 = ClusterNode('127.0.0.1', 7001) + RedisCluster(startup_nodes=[node_1, node_2]) + assert str(ex.value).startswith( + "startup_nodes could not agree on a valid slots cache"), str( + ex.value) + + def test_cluster_one_instance(self): + """ + If the cluster exists of only 1 node then there is some hacks that must + be validated they work. + """ + node = ClusterNode(default_host, default_port) + cluster_slots = [[0, 16383, ['', default_port]]] + rc = get_mocked_redis_client(startup_nodes=[node], + cluster_slots=cluster_slots) + + n = rc.nodes_manager + assert len(n.nodes_cache) == 1 + n_node = rc.get_node(node_name=node.name) + assert n_node is not None + assert n_node == node + assert n_node.server_type == PRIMARY + assert len(n.slots_cache) == REDIS_CLUSTER_HASH_SLOTS + for i in range(0, REDIS_CLUSTER_HASH_SLOTS): + assert n.slots_cache[i] == [n_node] + + def test_init_with_down_node(self): + """ + If I can't connect to one of the nodes, everything should still work. + But if I can't connect to any of the nodes, exception should be thrown. + """ + with patch.object(NodesManager, + 'create_redis_node') as create_redis_node: + def create_mocked_redis_node(host, port, **kwargs): + if port == 7000: + raise ConnectionError('mock connection error for 7000') + + r_node = Redis(host=host, port=port, decode_responses=True) + + def execute_command(*args, **kwargs): + if args[0] == 'CLUSTER SLOTS': + return [ + [ + 0, 8191, + ['127.0.0.1', 7001, 'node_1'], + ], + [ + 8192, 16383, + ['127.0.0.1', 7002, 'node_2'], + ] + ] + elif args[1] == 'cluster-require-full-coverage': + return {'cluster-require-full-coverage': 'yes'} + + r_node.execute_command = execute_command + + return r_node + + create_redis_node.side_effect = create_mocked_redis_node + + node_1 = ClusterNode('127.0.0.1', 7000) + node_2 = ClusterNode('127.0.0.1', 7001) + + # If all startup nodes fail to connect, connection error should be + # thrown + with pytest.raises(RedisClusterException) as e: + RedisCluster(startup_nodes=[node_1]) + assert 'Redis Cluster cannot be connected' in str(e.value) + + with patch.object(CommandsParser, 'initialize', + autospec=True) as cmd_parser_initialize: + + def cmd_init_mock(self, r): + self.commands = {'get': {'name': 'get', 'arity': 2, + 'flags': ['readonly', + 'fast'], + 'first_key_pos': 1, + 'last_key_pos': 1, + 'step_count': 1}} + + cmd_parser_initialize.side_effect = cmd_init_mock + # When at least one startup node is reachable, the cluster + # initialization should succeeds + rc = RedisCluster(startup_nodes=[node_1, node_2]) + assert rc.get_node(host=default_host, port=7001) is not None + assert rc.get_node(host=default_host, port=7002) is not None + + +@pytest.mark.onlycluster +class TestClusterPubSubObject: + """ + Tests for the ClusterPubSub class + """ + + def test_init_pubsub_with_host_and_port(self, r): + """ + Test creation of pubsub instance with passed host and port + """ + node = r.get_default_node() + p = r.pubsub(host=node.host, port=node.port) + assert p.get_pubsub_node() == node + + def test_init_pubsub_with_node(self, r): + """ + Test creation of pubsub instance with passed node + """ + node = r.get_default_node() + p = r.pubsub(node=node) + assert p.get_pubsub_node() == node + + def test_init_pubusub_without_specifying_node(self, r): + """ + Test creation of pubsub instance without specifying a node. The node + should be determined based on the keyslot of the first command + execution. + """ + channel_name = 'foo' + node = r.get_node_from_key(channel_name) + p = r.pubsub() + assert p.get_pubsub_node() is None + p.subscribe(channel_name) + assert p.get_pubsub_node() == node + + def test_init_pubsub_with_a_non_existent_node(self, r): + """ + Test creation of pubsub instance with node that doesn't exists in the + cluster. RedisClusterException should be raised. + """ + node = ClusterNode('1.1.1.1', 1111) + with pytest.raises(RedisClusterException): + r.pubsub(node) + + def test_init_pubsub_with_a_non_existent_host_port(self, r): + """ + Test creation of pubsub instance with host and port that don't belong + to a node in the cluster. + RedisClusterException should be raised. + """ + with pytest.raises(RedisClusterException): + r.pubsub(host='1.1.1.1', port=1111) + + def test_init_pubsub_host_or_port(self, r): + """ + Test creation of pubsub instance with host but without port, and vice + versa. DataError should be raised. + """ + with pytest.raises(DataError): + r.pubsub(host='localhost') + + with pytest.raises(DataError): + r.pubsub(port=16379) + + def test_get_redis_connection(self, r): + """ + Test that get_redis_connection() returns the redis connection of the + set pubsub node + """ + node = r.get_default_node() + p = r.pubsub(node=node) + assert p.get_redis_connection() == node.redis_connection + + +@pytest.mark.onlycluster +class TestClusterPipeline: + """ + Tests for the ClusterPipeline class + """ + + def test_blocked_methods(self, r): + """ + Currently some method calls on a Cluster pipeline + is blocked when using in cluster mode. + They maybe implemented in the future. + """ + pipe = r.pipeline() + with pytest.raises(RedisClusterException): + pipe.multi() + + with pytest.raises(RedisClusterException): + pipe.immediate_execute_command() + + with pytest.raises(RedisClusterException): + pipe._execute_transaction(None, None, None) + + with pytest.raises(RedisClusterException): + pipe.load_scripts() + + with pytest.raises(RedisClusterException): + pipe.watch() + + with pytest.raises(RedisClusterException): + pipe.unwatch() + + with pytest.raises(RedisClusterException): + pipe.script_load_for_pipeline(None) + + with pytest.raises(RedisClusterException): + pipe.eval() + + def test_blocked_arguments(self, r): + """ + Currently some arguments is blocked when using in cluster mode. + They maybe implemented in the future. + """ + with pytest.raises(RedisClusterException) as ex: + r.pipeline(transaction=True) + + assert str(ex.value).startswith( + "transaction is deprecated in cluster mode") is True + + with pytest.raises(RedisClusterException) as ex: + r.pipeline(shard_hint=True) + + assert str(ex.value).startswith( + "shard_hint is deprecated in cluster mode") is True + + def test_redis_cluster_pipeline(self, r): + """ + Test that we can use a pipeline with the RedisCluster class + """ + with r.pipeline() as pipe: + pipe.set("foo", "bar") + pipe.get("foo") + assert pipe.execute() == [True, b'bar'] + + def test_mget_disabled(self, r): + """ + Test that mget is disabled for ClusterPipeline + """ + with r.pipeline() as pipe: + with pytest.raises(RedisClusterException): + pipe.mget(['a']) + + def test_mset_disabled(self, r): + """ + Test that mset is disabled for ClusterPipeline + """ + with r.pipeline() as pipe: + with pytest.raises(RedisClusterException): + pipe.mset({'a': 1, 'b': 2}) + + def test_rename_disabled(self, r): + """ + Test that rename is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.rename('a', 'b') + + def test_renamenx_disabled(self, r): + """ + Test that renamenx is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.renamenx('a', 'b') + + def test_delete_single(self, r): + """ + Test a single delete operation + """ + r['a'] = 1 + with r.pipeline(transaction=False) as pipe: + pipe.delete('a') + assert pipe.execute() == [1] + + def test_multi_delete_unsupported(self, r): + """ + Test that multi delete operation is unsupported + """ + with r.pipeline(transaction=False) as pipe: + r['a'] = 1 + r['b'] = 2 + with pytest.raises(RedisClusterException): + pipe.delete('a', 'b') + + def test_brpoplpush_disabled(self, r): + """ + Test that brpoplpush is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.brpoplpush() + + def test_rpoplpush_disabled(self, r): + """ + Test that rpoplpush is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.rpoplpush() + + def test_sort_disabled(self, r): + """ + Test that sort is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sort() + + def test_sdiff_disabled(self, r): + """ + Test that sdiff is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sdiff() + + def test_sdiffstore_disabled(self, r): + """ + Test that sdiffstore is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sdiffstore() + + def test_sinter_disabled(self, r): + """ + Test that sinter is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sinter() + + def test_sinterstore_disabled(self, r): + """ + Test that sinterstore is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sinterstore() + + def test_smove_disabled(self, r): + """ + Test that move is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.smove() + + def test_sunion_disabled(self, r): + """ + Test that sunion is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sunion() + + def test_sunionstore_disabled(self, r): + """ + Test that sunionstore is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.sunionstore() + + def test_spfmerge_disabled(self, r): + """ + Test that spfmerge is disabled for ClusterPipeline + """ + with r.pipeline(transaction=False) as pipe: + with pytest.raises(RedisClusterException): + pipe.pfmerge() + + def test_multi_key_operation_with_a_single_slot(self, r): + """ + Test multi key operation with a single slot + """ + pipe = r.pipeline(transaction=False) + pipe.set('a{foo}', 1) + pipe.set('b{foo}', 2) + pipe.set('c{foo}', 3) + pipe.get('a{foo}') + pipe.get('b{foo}') + pipe.get('c{foo}') + + res = pipe.execute() + assert res == [True, True, True, b'1', b'2', b'3'] + + def test_multi_key_operation_with_multi_slots(self, r): + """ + Test multi key operation with more than one slot + """ + pipe = r.pipeline(transaction=False) + pipe.set('a{foo}', 1) + pipe.set('b{foo}', 2) + pipe.set('c{foo}', 3) + pipe.set('bar', 4) + pipe.set('bazz', 5) + pipe.get('a{foo}') + pipe.get('b{foo}') + pipe.get('c{foo}') + pipe.get('bar') + pipe.get('bazz') + res = pipe.execute() + assert res == [True, True, True, True, True, b'1', b'2', b'3', b'4', + b'5'] + + def test_connection_error_not_raised(self, r): + """ + Test that the pipeline doesn't raise an error on connection error when + raise_on_error=False + """ + key = 'foo' + node = r.get_node_from_key(key, False) + + def raise_connection_error(): + e = ConnectionError("error") + return e + + with r.pipeline() as pipe: + mock_node_resp_func(node, raise_connection_error) + res = pipe.get(key).get(key).execute(raise_on_error=False) + assert node.redis_connection.connection.read_response.called + assert isinstance(res[0], ConnectionError) + + def test_connection_error_raised(self, r): + """ + Test that the pipeline raises an error on connection error when + raise_on_error=True + """ + key = 'foo' + node = r.get_node_from_key(key, False) + + def raise_connection_error(): + e = ConnectionError("error") + return e + + with r.pipeline() as pipe: + mock_node_resp_func(node, raise_connection_error) + with pytest.raises(ConnectionError): + pipe.get(key).get(key).execute(raise_on_error=True) + + def test_asking_error(self, r): + """ + Test redirection on ASK error + """ + key = 'foo' + first_node = r.get_node_from_key(key, False) + ask_node = None + for node in r.get_nodes(): + if node != first_node: + ask_node = node + break + if ask_node is None: + warnings.warn("skipping this test since the cluster has only one " + "node") + return + ask_msg = "{0} {1}:{2}".format(r.keyslot(key), ask_node.host, + ask_node.port) + + def raise_ask_error(): + raise AskError(ask_msg) + + with r.pipeline() as pipe: + mock_node_resp_func(first_node, raise_ask_error) + mock_node_resp(ask_node, 'MOCK_OK') + res = pipe.get(key).execute() + assert first_node.redis_connection.connection.read_response.called + assert ask_node.redis_connection.connection.read_response.called + assert res == ['MOCK_OK'] + + def test_empty_stack(self, r): + """ + If pipeline is executed with no commands it should + return a empty list. + """ + p = r.pipeline() + result = p.execute() + assert result == [] + + +@pytest.mark.onlycluster +class TestReadOnlyPipeline: + """ + Tests for ClusterPipeline class in readonly mode + """ + + def test_pipeline_readonly(self, r): + """ + On readonly mode, we supports get related stuff only. + """ + r.readonly(target_nodes='all') + r.set('foo71', 'a1') # we assume this key is set on 127.0.0.1:7001 + r.zadd('foo88', + {'z1': 1}) # we assume this key is set on 127.0.0.1:7002 + r.zadd('foo88', {'z2': 4}) + + with r.pipeline() as readonly_pipe: + readonly_pipe.get('foo71').zrange('foo88', 0, 5, withscores=True) + assert readonly_pipe.execute() == [ + b'a1', + [(b'z1', 1.0), (b'z2', 4)], + ] + + def test_moved_redirection_on_slave_with_default(self, r): + """ + On Pipeline, we redirected once and finally get from master with + readonly client when data is completely moved. + """ + key = 'bar' + r.set(key, 'foo') + # set read_from_replicas to True + r.read_from_replicas = True + primary = r.get_node_from_key(key, False) + replica = r.get_node_from_key(key, True) + with r.pipeline() as readwrite_pipe: + mock_node_resp(primary, "MOCK_FOO") + if replica is not None: + moved_error = "{0} {1}:{2}".format(r.keyslot(key), + primary.host, + primary.port) + + def raise_moved_error(): + raise MovedError(moved_error) + + mock_node_resp_func(replica, raise_moved_error) + assert readwrite_pipe.reinitialize_counter == 0 + readwrite_pipe.get(key).get(key) + assert readwrite_pipe.execute() == ["MOCK_FOO", "MOCK_FOO"] + if replica is not None: + # the slot has a replica as well, so MovedError should have + # occurred. If MovedError occurs, we should see the + # reinitialize_counter increase. + assert readwrite_pipe.reinitialize_counter == 1 + conn = replica.redis_connection.connection + assert conn.read_response.called is True + + def test_readonly_pipeline_from_readonly_client(self, request): + """ + Test that the pipeline is initialized with readonly mode if the client + has it enabled + """ + # Create a cluster with reading from replications + ro = _get_client(RedisCluster, request, read_from_replicas=True) + key = 'bar' + ro.set(key, 'foo') + import time + time.sleep(0.2) + with ro.pipeline() as readonly_pipe: + mock_all_nodes_resp(ro, 'MOCK_OK') + assert readonly_pipe.read_from_replicas is True + assert readonly_pipe.get(key).get( + key).execute() == ['MOCK_OK', 'MOCK_OK'] + slot_nodes = ro.nodes_manager.slots_cache[ro.keyslot(key)] + if len(slot_nodes) > 1: + executed_on_replica = False + for node in slot_nodes: + if node.server_type == REPLICA: + conn = node.redis_connection.connection + executed_on_replica = conn.read_response.called + if executed_on_replica: + break + assert executed_on_replica is True -- cgit v1.2.1