<s> from pymemcache . client . hash import HashClient <EOL> from pymemcache . client . base import Client , PooledClient <EOL> from pymemcache . exceptions import MemcacheUnknownError <EOL> from pymemcache import pool <EOL> from . test_client import ClientTestMixin , MockSocket <EOL> import unittest <EOL> import pytest <EOL> import mock <EOL> import socket <EOL> class TestHashClient ( ClientTestMixin , unittest . TestCase ) : <EOL> def make_client_pool ( self , hostname , mock_socket_values , serializer = None ) : <EOL> mock_client = Client ( hostname , serializer = serializer ) <EOL> mock_client . sock = MockSocket ( mock_socket_values ) <EOL> client = PooledClient ( hostname , serializer = serializer ) <EOL> client . client_pool = pool . ObjectPool ( lambda : mock_client ) <EOL> return mock_client <EOL> def make_client ( self , * mock_socket_values , ** kwargs ) : <EOL> current_port = <NUM_LIT> <EOL> client = HashClient ( [ ] , ** kwargs ) <EOL> ip = '<STR_LIT:127.0.0.1>' <EOL> for vals in mock_socket_values : <EOL> s = '<STR_LIT>' % ( ip , current_port ) <EOL> c = self . make_client_pool ( <EOL> ( ip , current_port ) , <EOL> vals <EOL> ) <EOL> client . clients [ s ] = c <EOL> client . hasher . add_node ( s ) <EOL> current_port += <NUM_LIT:1> <EOL> return client <EOL> def test_setup_client_without_pooling ( self ) : <EOL> with mock . patch ( '<STR_LIT>' ) as internal_client : <EOL> client = HashClient ( [ ] , timeout = <NUM_LIT> , key_prefix = '<STR_LIT>' ) <EOL> client . add_server ( '<STR_LIT:127.0.0.1>' , '<STR_LIT>' ) <EOL> assert internal_client . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] == ( '<STR_LIT:127.0.0.1>' , '<STR_LIT>' ) <EOL> kwargs = internal_client . call_args [ <NUM_LIT:1> ] <EOL> assert kwargs [ '<STR_LIT>' ] == <NUM_LIT> <EOL> assert kwargs [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> def test_get_many_all_found ( self ) : <EOL> client = self . make_client ( * [ <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> ] ) <EOL> def get_clients ( key ) : <EOL> if key == b'<STR_LIT>' : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> else : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> client . _get_client = get_clients <EOL> result = client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> result = client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> result = client . get_many ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> assert result == { b'<STR_LIT>' : b'<STR_LIT>' , b'<STR_LIT>' : b'<STR_LIT>' } <EOL> def test_get_many_some_found ( self ) : <EOL> client = self . make_client ( * [ <EOL> [ b'<STR_LIT>' , ] , <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> ] ) <EOL> def get_clients ( key ) : <EOL> if key == b'<STR_LIT>' : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> else : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> client . _get_client = get_clients <EOL> result = client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> result = client . get_many ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> assert result == { b'<STR_LIT>' : b'<STR_LIT>' } <EOL> def test_get_many_bad_server_data ( self ) : <EOL> client = self . make_client ( * [ <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> ] ) <EOL> def get_clients ( key ) : <EOL> if key == b'<STR_LIT>' : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> else : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> client . _get_client = get_clients <EOL> with pytest . raises ( MemcacheUnknownError ) : <EOL> client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> client . get_many ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> def test_get_many_bad_server_data_ignore ( self ) : <EOL> client = self . make_client ( * [ <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> [ b'<STR_LIT>' , b'<STR_LIT>' , ] , <EOL> ] , ignore_exc = True ) <EOL> def get_clients ( key ) : <EOL> if key == b'<STR_LIT>' : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> else : <EOL> return client . clients [ '<STR_LIT>' ] <EOL> client . _get_client = get_clients <EOL> client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> client . set ( b'<STR_LIT>' , b'<STR_LIT>' , noreply = False ) <EOL> result = client . get_many ( [ b'<STR_LIT>' , b'<STR_LIT>' ] ) <EOL> assert result == { } <EOL> def test_no_servers_left ( self ) : <EOL> from pymemcache . client . hash import HashClient <EOL> client = HashClient ( <EOL> [ ] , use_pooling = True , <EOL> ignore_exc = True , <EOL> timeout = <NUM_LIT:1> , connect_timeout = <NUM_LIT:1> <EOL> ) <EOL> hashed_client = client . _get_client ( '<STR_LIT:foo>' ) <EOL> assert hashed_client is None <EOL> def test_no_servers_left_raise_exception ( self ) : <EOL> from pymemcache . client . hash import HashClient <EOL> client = HashClient ( <EOL> [ ] , use_pooling = True , <EOL> ignore_exc = False , <EOL> timeout = <NUM_LIT:1> , connect_timeout = <NUM_LIT:1> <EOL> ) <EOL> with pytest . raises ( Exception ) as e : <EOL> client . _get_client ( '<STR_LIT:foo>' ) <EOL> assert str ( e . value ) == '<STR_LIT>' <EOL> def test_unavailable_servers_zero_retry_raise_exception ( self ) : <EOL> from pymemcache . client . hash import HashClient <EOL> client = HashClient ( <EOL> [ ( '<STR_LIT>' , <NUM_LIT> ) ] , use_pooling = True , <EOL> ignore_exc = False , <EOL> retry_attempts = <NUM_LIT:0> , timeout = <NUM_LIT:1> , connect_timeout = <NUM_LIT:1> <EOL> ) <EOL> with pytest . raises ( socket . error ) as e : <EOL> client . get ( '<STR_LIT:foo>' ) <EOL> def test_no_servers_left_with_commands ( self ) : <EOL> from pymemcache . client . hash import HashClient <EOL> client = HashClient ( <EOL> [ ] , use_pooling = True , <EOL> ignore_exc = True , <EOL> timeout = <NUM_LIT:1> , connect_timeout = <NUM_LIT:1> <EOL> ) <EOL> result = client . get ( '<STR_LIT:foo>' ) <EOL> assert result is False <EOL> def test_no_servers_left_with_set_many ( self ) : <EOL> from pymemcache . client . hash import HashClient <EOL> client = HashClient ( <EOL> [ ] , use_pooling = True , <EOL> ignore_exc = True , <EOL> timeout = <NUM_LIT:1> , connect_timeout = <NUM_LIT:1> <EOL> ) <EOL> result = client . set_many ( { '<STR_LIT:foo>' : '<STR_LIT:bar>' } ) <EOL> assert result is False <EOL> def test_no_servers_left_with_get_many ( self ) : <EOL> from pymemcache . client . hash import HashClient <EOL> client = HashClient ( <EOL> [ ] , use_pooling = True , <EOL> ignore_exc = True , <EOL> timeout = <NUM_LIT:1> , connect_timeout = <NUM_LIT:1> <EOL> ) <EOL> result = client . get_many ( [ '<STR_LIT:foo>' , '<STR_LIT:bar>' ] ) <EOL> assert result == { '<STR_LIT:foo>' : False , '<STR_LIT:bar>' : False } </s>
