<s> import pytest <EOL> import redis <EOL> from mock import Mock <EOL> from distutils . version import StrictVersion <EOL> _REDIS_VERSIONS = { } <EOL> def get_version ( ** kwargs ) : <EOL> params = { '<STR_LIT:host>' : '<STR_LIT:localhost>' , '<STR_LIT:port>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:9> } <EOL> params . update ( kwargs ) <EOL> key = '<STR_LIT>' % ( params [ '<STR_LIT:host>' ] , params [ '<STR_LIT:port>' ] ) <EOL> if key not in _REDIS_VERSIONS : <EOL> client = redis . Redis ( ** params ) <EOL> _REDIS_VERSIONS [ key ] = client . info ( ) [ '<STR_LIT>' ] <EOL> client . connection_pool . disconnect ( ) <EOL> return _REDIS_VERSIONS [ key ] <EOL> def _get_client ( cls , request = None , ** kwargs ) : <EOL> params = { '<STR_LIT:host>' : '<STR_LIT:localhost>' , '<STR_LIT:port>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:9> } <EOL> params . update ( kwargs ) <EOL> client = cls ( ** params ) <EOL> client . flushdb ( ) <EOL> if request : <EOL> def teardown ( ) : <EOL> client . flushdb ( ) <EOL> client . connection_pool . disconnect ( ) <EOL> request . addfinalizer ( teardown ) <EOL> return client <EOL> def skip_if_server_version_lt ( min_version ) : <EOL> check = StrictVersion ( get_version ( ) ) < StrictVersion ( min_version ) <EOL> return pytest . mark . skipif ( check , reason = "<STR_LIT>" ) <EOL> @ pytest . fixture ( ) <EOL> def r ( request , ** kwargs ) : <EOL> return _get_client ( redis . Redis , request , ** kwargs ) <EOL> @ pytest . fixture ( ) <EOL> def sr ( request , ** kwargs ) : <EOL> return _get_client ( redis . StrictRedis , request , ** kwargs ) <EOL> def _gen_cluster_mock_resp ( r , response ) : <EOL> mock_connection_pool = Mock ( ) <EOL> connection = Mock ( ) <EOL> response = response <EOL> connection . read_response . return_value = response <EOL> mock_connection_pool . get_connection . return_value = connection <EOL> r . connection_pool = mock_connection_pool <EOL> return r <EOL> @ pytest . fixture ( ) <EOL> def mock_cluster_resp_ok ( request , ** kwargs ) : <EOL> r = _get_client ( redis . Redis , request , ** kwargs ) <EOL> return _gen_cluster_mock_resp ( r , '<STR_LIT:OK>' ) <EOL> @ pytest . fixture ( ) <EOL> def mock_cluster_resp_int ( request , ** kwargs ) : <EOL> r = _get_client ( redis . Redis , request , ** kwargs ) <EOL> return _gen_cluster_mock_resp ( r , '<STR_LIT:2>' ) <EOL> @ pytest . fixture ( ) <EOL> def mock_cluster_resp_info ( request , ** kwargs ) : <EOL> r = _get_client ( redis . Redis , request , ** kwargs ) <EOL> response = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return _gen_cluster_mock_resp ( r , response ) <EOL> @ pytest . fixture ( ) <EOL> def mock_cluster_resp_nodes ( request , ** kwargs ) : <EOL> r = _get_client ( redis . Redis , request , ** kwargs ) <EOL> response = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> return _gen_cluster_mock_resp ( r , response ) <EOL> @ pytest . fixture ( ) <EOL> def mock_cluster_resp_slaves ( request , ** kwargs ) : <EOL> r = _get_client ( redis . Redis , request , ** kwargs ) <EOL> response = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> return _gen_cluster_mock_resp ( r , response ) </s>
