本文整理汇总了Python中rediscluster.nodemanager.NodeManager类的典型用法代码示例。如果您正苦于以下问题:Python NodeManager类的具体用法?Python NodeManager怎么用?Python NodeManager使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了NodeManager类的17个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_cluster_one_instance
def test_cluster_one_instance():
"""
If the cluster exists of only 1 node then there is some hacks that must
be validated they work.
"""
with patch.object(StrictRedis, 'execute_command') as mock_execute_command:
return_data = [[0, 16383, ['', 7006]]]
def patch_execute_command(*args, **kwargs):
if args == ('CONFIG GET', 'cluster-require-full-coverage'):
return {'cluster-require-full-coverage': 'yes'}
else:
return return_data
# mock_execute_command.return_value = return_data
mock_execute_command.side_effect = patch_execute_command
n = NodeManager(startup_nodes=[{"host": "127.0.0.1", "port": 7006}])
n.initialize()
assert n.nodes == {"127.0.0.1:7006": {
'host': '127.0.0.1',
'name': '127.0.0.1:7006',
'port': 7006,
'server_type': 'master',
}}
assert len(n.slots) == 16384
for i in range(0, 16384):
assert n.slots[i] == [{
"host": "127.0.0.1",
"name": "127.0.0.1:7006",
"port": 7006,
"server_type": "master",
}]
开发者ID:Grokzen,项目名称:redis-py-cluster,代码行数:35,代码来源:test_node_manager.py
示例2: test_cluster_one_instance
def test_cluster_one_instance():
"""
If the cluster exists of only 1 node then there is some hacks that must
be validated they work.
"""
with patch.object(StrictRedis, 'execute_command') as mock_execute_command:
return_data = [[0, 16383, ['', 7006]]]
mock_execute_command.return_value = return_data
n = NodeManager(startup_nodes=[{"host": "127.0.0.1", "port": 7006}])
n.initialize()
assert n.nodes == {"127.0.0.1:7006": {
'host': '127.0.0.1',
'name': '127.0.0.1:7006',
'port': 7006,
'server_type': 'master',
}}
assert len(n.slots) == 16384
assert n.slots[0] == {
"host": "127.0.0.1",
"name": "127.0.0.1:7006",
"port": 7006,
"server_type": "master",
}
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:26,代码来源:test_node_manager.py
示例3: test_reset
def test_reset():
"""
Test that reset method resets variables back to correct default values.
"""
n = NodeManager(startup_nodes=[{}])
n.initialize = Mock()
n.reset()
assert n.initialize.call_count == 1
开发者ID:Grokzen,项目名称:redis-py-cluster,代码行数:8,代码来源:test_node_manager.py
示例4: test_random_startup_node
def test_random_startup_node():
"""
Hard to test reliable for a random
"""
s = [{"1": 1}, {"2": 2}, {"3": 3}],
n = NodeManager(startup_nodes=s)
random_node = n.random_startup_node()
for i in range(0, 5):
assert random_node in s
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:10,代码来源:test_node_manager.py
示例5: test_keyslot
def test_keyslot():
"""
Test that method will compute correct key in all supported cases
"""
n = NodeManager([{}])
assert n.keyslot("foo") == 12182
assert n.keyslot("{foo}bar") == 12182
assert n.keyslot("{foo}") == 12182
assert n.keyslot(1337) == 4314
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:10,代码来源:test_node_manager.py
示例6: test_random_startup_node_ittr
def test_random_startup_node_ittr():
"""
Hard to test reliable for a random function
"""
s = [{"1": 1}, {"2": 2}, {"3": 3}],
n = NodeManager(startup_nodes=s)
for i, node in enumerate(n.random_startup_node_ittr()):
if i == 5:
break
assert node in s
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:11,代码来源:test_node_manager.py
示例7: test_all_nodes
def test_all_nodes():
"""
Set a list of nodes and it should be possible to itterate over all
"""
n = NodeManager(startup_nodes=[{"host": "127.0.0.1", "port": 7000}])
n.initialize()
nodes = [node for node in n.nodes.values()]
for i, node in enumerate(n.all_nodes()):
assert node in nodes
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:11,代码来源:test_node_manager.py
示例8: test_all_nodes_masters
def test_all_nodes_masters():
"""
Set a list of nodes with random masters/slaves config and it shold be possible
to itterate over all of them.
"""
n = NodeManager(startup_nodes=[{"host": "127.0.0.1", "port": 7000}, {"host": "127.0.0.1", "port": 7001}])
n.initialize()
nodes = [node for node in n.nodes.values() if node['server_type'] == 'master']
for node in n.all_masters():
assert node in nodes
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:12,代码来源:test_node_manager.py
示例9: test_cluster_slots_error
def test_cluster_slots_error():
"""
Check that exception is raised if initialize can't execute
'CLUSTER SLOTS' command.
"""
with patch.object(StrictRedisCluster, 'execute_command') as execute_command_mock:
execute_command_mock.side_effect = Exception("foobar")
n = NodeManager(startup_nodes=[{}])
with pytest.raises(RedisClusterException):
n.initialize()
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:12,代码来源:test_node_manager.py
示例10: test_determine_pubsub_node
def test_determine_pubsub_node():
"""
Given a set of nodes it should determine the same pubsub node each time.
"""
n = NodeManager(startup_nodes=[{}])
n.nodes = {
"127.0.0.1:7001": {"host": "127.0.0.1", "port": 7001, "server_type": "master"},
"127.0.0.1:7005": {"host": "127.0.0.1", "port": 7005, "server_type": "master"},
"127.0.0.1:7000": {"host": "127.0.0.1", "port": 7000, "server_type": "master"},
"127.0.0.1:7002": {"host": "127.0.0.1", "port": 7002, "server_type": "master"},
}
n.determine_pubsub_node()
assert n.pubsub_node == {"host": "127.0.0.1", "port": 7005, "server_type": "master", "pubsub": True}
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:15,代码来源:test_node_manager.py
示例11: test_init_with_down_node
def test_init_with_down_node():
"""
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.
"""
def get_redis_link(host, port, decode_responses=False):
if port == 7000:
raise ConnectionError('mock connection error for 7000')
return StrictRedis(host=host, port=port, decode_responses=decode_responses)
with patch.object(NodeManager, 'get_redis_link', side_effect=get_redis_link):
n = NodeManager(startup_nodes=[{"host": "127.0.0.1", "port": 7000}])
with pytest.raises(RedisClusterException) as e:
n.initialize()
assert 'Redis Cluster cannot be connected' in unicode(e.value)
开发者ID:Grokzen,项目名称:redis-py-cluster,代码行数:15,代码来源:test_node_manager.py
示例12: test_set_node
def test_set_node():
"""
Test to update data in a slot.
"""
expected = {
"host": "127.0.0.1",
"name": "127.0.0.1:7000",
"port": 7000,
"server_type": "master",
}
n = NodeManager(startup_nodes=[{}])
assert len(n.slots) == 0, "no slots should exist"
res = n.set_node(host="127.0.0.1", port=7000, server_type="master")
assert res == expected
assert n.nodes == {expected['name']: expected}
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:16,代码来源:test_node_manager.py
示例13: test_init_slots_cache_slots_collision
def test_init_slots_cache_slots_collision():
"""
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.
"""
n = NodeManager(startup_nodes=[
{"host": "127.0.0.1", "port": 7000},
{"host": "127.0.0.1", "port": 7001},
])
def monkey_link(host=None, port=None, *args, **kwargs):
"""
Helper function to return custom slots cache data from different redis nodes
"""
if port == 7000:
result = [[0, 5460, [b'127.0.0.1', 7000], [b'127.0.0.1', 7003]],
[5461, 10922, [b'127.0.0.1', 7001], [b'127.0.0.1', 7004]]]
elif port == 7001:
result = [[0, 5460, [b'127.0.0.1', 7001], [b'127.0.0.1', 7003]],
[5461, 10922, [b'127.0.0.1', 7000], [b'127.0.0.1', 7004]]]
else:
result = []
r = StrictRedisCluster(host=host, port=port, decode_responses=True)
orig_execute_command = r.execute_command
def execute_command(*args, **kwargs):
if args == ("cluster", "slots"):
return result
elif args == ('CONFIG GET', 'cluster-require-full-coverage'):
return {'cluster-require-full-coverage': 'yes'}
else:
return orig_execute_command(*args, **kwargs)
r.execute_command = execute_command
return r
n.get_redis_link = monkey_link
with pytest.raises(RedisClusterException) as ex:
n.initialize()
assert unicode(ex.value).startswith("startup_nodes could not agree on a valid slots cache."), unicode(ex.value)
开发者ID:Grokzen,项目名称:redis-py-cluster,代码行数:45,代码来源:test_node_manager.py
示例14: test_flush_slots_nodes_cache
def test_flush_slots_nodes_cache():
"""
Slots cache should already be populated.
"""
n = NodeManager([{"host": "127.0.0.1", "port": 7000}])
n.initialize()
assert len(n.slots) == NodeManager.RedisClusterHashSlots
assert len(n.nodes) == 6
n.flush_slots_cache()
n.flush_nodes_cache()
assert len(n.slots) == 0
assert len(n.nodes) == 0
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:14,代码来源:test_node_manager.py
示例15: test_reset
def test_reset():
"""
Test that reset method resets variables back to correct default values.
"""
n = NodeManager(startup_nodes=[{}])
n.initialize = Mock()
n.slots = {"foo": "bar"}
n.nodes = ["foo", "bar"]
n.reset()
assert n.slots == {}
assert n.nodes == {}
开发者ID:huangfupeng,项目名称:redis-py-cluster,代码行数:12,代码来源:test_node_manager.py
示例16: test_keyslot
def test_keyslot():
"""
Test that method will compute correct key in all supported cases
"""
n = NodeManager([{}])
assert n.keyslot("foo") == 12182
assert n.keyslot("{foo}bar") == 12182
assert n.keyslot("{foo}") == 12182
assert n.keyslot(1337) == 4314
assert n.keyslot(125) == n.keyslot(b"125")
assert n.keyslot(125) == n.keyslot("\x31\x32\x35")
assert n.keyslot("大奖") == n.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96")
assert n.keyslot(u"大奖") == n.keyslot(b"\xe5\xa4\xa7\xe5\xa5\x96")
assert n.keyslot(1337.1234) == n.keyslot("1337.1234")
assert n.keyslot(1337) == n.keyslot("1337")
assert n.keyslot(b"abc") == n.keyslot("abc")
assert n.keyslot("abc") == n.keyslot(unicode("abc"))
assert n.keyslot(unicode("abc")) == n.keyslot(b"abc")
开发者ID:Grokzen,项目名称:redis-py-cluster,代码行数:20,代码来源:test_node_manager.py
示例17: test_initialize_follow_cluster
def test_initialize_follow_cluster():
n = NodeManager(nodemanager_follow_cluster=True, startup_nodes=[{'host': '127.0.0.1', 'port': 7000}])
n.orig_startup_nodes = None
n.initialize()
开发者ID:Grokzen,项目名称:redis-py-cluster,代码行数:4,代码来源:test_node_manager.py
注:本文中的rediscluster.nodemanager.NodeManager类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论