/usr/lib/python2.7/dist-packages/pymemcache/test/test_client_hash.py is in python-pymemcache 1.3.2-2ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | from pymemcache.client.hash import HashClient
from pymemcache.client.base import Client, PooledClient
from pymemcache.exceptions import MemcacheUnknownError
from pymemcache import pool
from .test_client import ClientTestMixin, MockSocket
import unittest
import pytest
import mock
class TestHashClient(ClientTestMixin, unittest.TestCase):
def make_client_pool(self, hostname, mock_socket_values, serializer=None):
mock_client = Client(hostname, serializer=serializer)
mock_client.sock = MockSocket(mock_socket_values)
client = PooledClient(hostname, serializer=serializer)
client.client_pool = pool.ObjectPool(lambda: mock_client)
return mock_client
def make_client(self, *mock_socket_values, **kwargs):
current_port = 11012
client = HashClient([], **kwargs)
ip = '127.0.0.1'
for vals in mock_socket_values:
s = '%s:%s' % (ip, current_port)
c = self.make_client_pool(
(ip, current_port),
vals
)
client.clients[s] = c
client.hasher.add_node(s)
current_port += 1
return client
def test_setup_client_without_pooling(self):
with mock.patch('pymemcache.client.hash.Client') as internal_client:
client = HashClient([], timeout=999, key_prefix='foo_bar_baz')
client.add_server('127.0.0.1', '11211')
assert internal_client.call_args[0][0] == ('127.0.0.1', '11211')
kwargs = internal_client.call_args[1]
assert kwargs['timeout'] == 999
assert kwargs['key_prefix'] == 'foo_bar_baz'
def test_get_many_all_found(self):
client = self.make_client(*[
[b'STORED\r\n', b'VALUE key3 0 6\r\nvalue2\r\nEND\r\n', ],
[b'STORED\r\n', b'VALUE key1 0 6\r\nvalue1\r\nEND\r\n', ],
])
def get_clients(key):
if key == b'key3':
return client.clients['127.0.0.1:11012']
else:
return client.clients['127.0.0.1:11013']
client._get_client = get_clients
result = client.set(b'key1', b'value1', noreply=False)
result = client.set(b'key3', b'value2', noreply=False)
result = client.get_many([b'key1', b'key3'])
assert result == {b'key1': b'value1', b'key3': b'value2'}
def test_get_many_some_found(self):
client = self.make_client(*[
[b'END\r\n', ],
[b'STORED\r\n', b'VALUE key1 0 6\r\nvalue1\r\nEND\r\n', ],
])
def get_clients(key):
if key == b'key3':
return client.clients['127.0.0.1:11012']
else:
return client.clients['127.0.0.1:11013']
client._get_client = get_clients
result = client.set(b'key1', b'value1', noreply=False)
result = client.get_many([b'key1', b'key3'])
assert result == {b'key1': b'value1'}
def test_get_many_bad_server_data(self):
client = self.make_client(*[
[b'STORED\r\n', b'VAXLUE key3 0 6\r\nvalue2\r\nEND\r\n', ],
[b'STORED\r\n', b'VAXLUE key1 0 6\r\nvalue1\r\nEND\r\n', ],
])
def get_clients(key):
if key == b'key3':
return client.clients['127.0.0.1:11012']
else:
return client.clients['127.0.0.1:11013']
client._get_client = get_clients
with pytest.raises(MemcacheUnknownError):
client.set(b'key1', b'value1', noreply=False)
client.set(b'key3', b'value2', noreply=False)
client.get_many([b'key1', b'key3'])
def test_get_many_bad_server_data_ignore(self):
client = self.make_client(*[
[b'STORED\r\n', b'VAXLUE key3 0 6\r\nvalue2\r\nEND\r\n', ],
[b'STORED\r\n', b'VAXLUE key1 0 6\r\nvalue1\r\nEND\r\n', ],
], ignore_exc=True)
def get_clients(key):
if key == b'key3':
return client.clients['127.0.0.1:11012']
else:
return client.clients['127.0.0.1:11013']
client._get_client = get_clients
client.set(b'key1', b'value1', noreply=False)
client.set(b'key3', b'value2', noreply=False)
result = client.get_many([b'key1', b'key3'])
assert result == {}
# TODO: Test failover logic
|