|
12 | 12 |
|
13 | 13 | import pytest
|
14 | 14 |
|
| 15 | +import redis.cluster |
15 | 16 | from redis import Redis
|
16 | 17 | from redis.backoff import (
|
17 | 18 | ConstantBackoff,
|
@@ -3048,6 +3049,33 @@ def raise_ask_error():
|
3048 | 3049 | assert ask_node.redis_connection.connection.read_response.called
|
3049 | 3050 | assert res == ["MOCK_OK"]
|
3050 | 3051 |
|
| 3052 | + @pytest.mark.parametrize("error", [ConnectionError, TimeoutError]) |
| 3053 | + def test_return_previous_acquired_connections(self, r, error): |
| 3054 | + # in order to ensure that a pipeline will make use of connections |
| 3055 | + # from different nodes |
| 3056 | + assert r.keyslot("a") != r.keyslot("b") |
| 3057 | + |
| 3058 | + orig_func = redis.cluster.get_connection |
| 3059 | + with patch("redis.cluster.get_connection") as get_connection: |
| 3060 | + |
| 3061 | + def raise_error(target_node, *args, **kwargs): |
| 3062 | + if get_connection.call_count == 2: |
| 3063 | + raise error("mocked error") |
| 3064 | + else: |
| 3065 | + return orig_func(target_node, *args, **kwargs) |
| 3066 | + |
| 3067 | + get_connection.side_effect = raise_error |
| 3068 | + |
| 3069 | + r.pipeline().get("a").get("b").execute() |
| 3070 | + |
| 3071 | + # there should have been two get_connections per execution and |
| 3072 | + # two executions due to exception raised in the first execution |
| 3073 | + assert get_connection.call_count == 4 |
| 3074 | + for cluster_node in r.nodes_manager.nodes_cache.values(): |
| 3075 | + connection_pool = cluster_node.redis_connection.connection_pool |
| 3076 | + num_of_conns = len(connection_pool._available_connections) |
| 3077 | + assert num_of_conns == connection_pool._created_connections |
| 3078 | + |
3051 | 3079 | def test_empty_stack(self, r):
|
3052 | 3080 | """
|
3053 | 3081 | If pipeline is executed with no commands it should
|
|
0 commit comments