<s> import struct , time , logging , sys , traceback <EOL> from cassandra import ConsistencyLevel , Unavailable , OperationTimedOut , ReadTimeout , ReadFailure , WriteTimeout , WriteFailure <EOL> from cassandra . cluster import Cluster , NoHostAvailable , Session <EOL> from cassandra . concurrent import execute_concurrent_with_args <EOL> from cassandra . metadata import murmur3 <EOL> from cassandra . policies import ( RoundRobinPolicy , DCAwareRoundRobinPolicy , <EOL> TokenAwarePolicy , WhiteListRoundRobinPolicy ) <EOL> from cassandra . query import SimpleStatement <EOL> from tests . integration import use_singledc , use_multidc , remove_cluster , PROTOCOL_VERSION <EOL> from tests . integration . long . utils import ( wait_for_up , create_schema , <EOL> CoordinatorStats , force_stop , <EOL> wait_for_down , decommission , start , <EOL> bootstrap , stop , IP_FORMAT ) <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> log = logging . getLogger ( __name__ ) <EOL> class LoadBalancingPolicyTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> remove_cluster ( ) <EOL> self . coordinator_stats = CoordinatorStats ( ) <EOL> self . prepared = None <EOL> @ classmethod <EOL> def teardown_class ( cls ) : <EOL> remove_cluster ( ) <EOL> def _insert ( self , session , keyspace , count = <NUM_LIT:12> , <EOL> consistency_level = ConsistencyLevel . ONE ) : <EOL> session . execute ( '<STR_LIT>' % keyspace ) <EOL> ss = SimpleStatement ( '<STR_LIT>' , consistency_level = consistency_level ) <EOL> tries = <NUM_LIT:0> <EOL> while tries < <NUM_LIT:100> : <EOL> try : <EOL> execute_concurrent_with_args ( session , ss , [ None ] * count ) <EOL> return <EOL> except ( OperationTimedOut , WriteTimeout , WriteFailure ) : <EOL> ex_type , ex , tb = sys . exc_info ( ) <EOL> log . warn ( "<STR_LIT>" . format ( ex_type . __name__ , ex , traceback . extract_tb ( tb ) ) ) <EOL> del tb <EOL> tries += <NUM_LIT:1> <EOL> raise RuntimeError ( "<STR_LIT>" . format ( ss ) ) <EOL> def _query ( self , session , keyspace , count = <NUM_LIT:12> , <EOL> consistency_level = ConsistencyLevel . ONE , use_prepared = False ) : <EOL> if use_prepared : <EOL> query_string = '<STR_LIT>' % keyspace <EOL> if not self . prepared or self . prepared . query_string != query_string : <EOL> self . prepared = session . prepare ( query_string ) <EOL> self . prepared . consistency_level = consistency_level <EOL> for i in range ( count ) : <EOL> tries = <NUM_LIT:0> <EOL> while True : <EOL> if tries > <NUM_LIT:100> : <EOL> raise RuntimeError ( "<STR_LIT>" . format ( self . prepared ) ) <EOL> try : <EOL> self . coordinator_stats . add_coordinator ( session . execute_async ( self . prepared . bind ( ( <NUM_LIT:0> , ) ) ) ) <EOL> break <EOL> except ( OperationTimedOut , ReadTimeout , ReadFailure ) : <EOL> ex_type , ex , tb = sys . exc_info ( ) <EOL> log . warn ( "<STR_LIT>" . format ( ex_type . __name__ , ex , traceback . extract_tb ( tb ) ) ) <EOL> del tb <EOL> tries += <NUM_LIT:1> <EOL> else : <EOL> routing_key = struct . pack ( '<STR_LIT>' , <NUM_LIT:0> ) <EOL> for i in range ( count ) : <EOL> ss = SimpleStatement ( '<STR_LIT>' % keyspace , <EOL> consistency_level = consistency_level , <EOL> routing_key = routing_key ) <EOL> tries = <NUM_LIT:0> <EOL> while True : <EOL> if tries > <NUM_LIT:100> : <EOL> raise RuntimeError ( "<STR_LIT>" . format ( ss ) ) <EOL> try : <EOL> self . coordinator_stats . add_coordinator ( session . execute_async ( ss ) ) <EOL> break <EOL> except ( OperationTimedOut , ReadTimeout , ReadFailure ) : <EOL> ex_type , ex , tb = sys . exc_info ( ) <EOL> log . warn ( "<STR_LIT>" . format ( ex_type . __name__ , ex , traceback . extract_tb ( tb ) ) ) <EOL> del tb <EOL> tries += <NUM_LIT:1> <EOL> def test_token_aware_is_used_by_default ( self ) : <EOL> """<STR_LIT>""" <EOL> cluster = Cluster ( protocol_version = PROTOCOL_VERSION ) <EOL> if murmur3 is not None : <EOL> self . assertTrue ( isinstance ( cluster . load_balancing_policy , TokenAwarePolicy ) ) <EOL> else : <EOL> self . assertTrue ( isinstance ( cluster . load_balancing_policy , DCAwareRoundRobinPolicy ) ) <EOL> cluster . shutdown ( ) <EOL> def test_roundrobin ( self ) : <EOL> use_singledc ( ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = RoundRobinPolicy ( ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> ) <EOL> create_schema ( cluster , session , keyspace , replication_factor = <NUM_LIT:3> ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:4> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:4> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> force_stop ( <NUM_LIT:3> ) <EOL> wait_for_down ( cluster , <NUM_LIT:3> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> decommission ( <NUM_LIT:1> ) <EOL> start ( <NUM_LIT:3> ) <EOL> wait_for_down ( cluster , <NUM_LIT:1> ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:6> ) <EOL> def test_roundrobin_two_dcs ( self ) : <EOL> use_multidc ( [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = RoundRobinPolicy ( ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:4> ) <EOL> create_schema ( cluster , session , keyspace , replication_strategy = [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:3> ) <EOL> force_stop ( <NUM_LIT:1> ) <EOL> bootstrap ( <NUM_LIT:5> , '<STR_LIT>' ) <EOL> self . _insert ( session , keyspace , count = <NUM_LIT:1000> ) <EOL> wait_for_up ( cluster , <NUM_LIT:5> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:3> ) <EOL> cluster . shutdown ( ) <EOL> def test_roundrobin_two_dcs_2 ( self ) : <EOL> use_multidc ( [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = RoundRobinPolicy ( ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:4> ) <EOL> create_schema ( cluster , session , keyspace , replication_strategy = [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:3> ) <EOL> force_stop ( <NUM_LIT:1> ) <EOL> bootstrap ( <NUM_LIT:5> , '<STR_LIT>' ) <EOL> self . _insert ( session , keyspace , count = <NUM_LIT:1000> ) <EOL> wait_for_up ( cluster , <NUM_LIT:5> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:3> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:3> ) <EOL> cluster . shutdown ( ) <EOL> def test_dc_aware_roundrobin_two_dcs ( self ) : <EOL> use_multidc ( [ <NUM_LIT:3> , <NUM_LIT:2> ] ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = DCAwareRoundRobinPolicy ( '<STR_LIT>' ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:4> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:5> ) <EOL> create_schema ( cluster , session , keyspace , replication_strategy = [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:4> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:4> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:4> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:0> ) <EOL> cluster . shutdown ( ) <EOL> def test_dc_aware_roundrobin_two_dcs_2 ( self ) : <EOL> use_multidc ( [ <NUM_LIT:3> , <NUM_LIT:2> ] ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = DCAwareRoundRobinPolicy ( '<STR_LIT>' ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:4> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:5> ) <EOL> create_schema ( cluster , session , keyspace , replication_strategy = [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:6> ) <EOL> cluster . shutdown ( ) <EOL> def test_dc_aware_roundrobin_one_remote_host ( self ) : <EOL> use_multidc ( [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = DCAwareRoundRobinPolicy ( '<STR_LIT>' , used_hosts_per_remote_dc = <NUM_LIT:1> ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:4> ) <EOL> create_schema ( cluster , session , keyspace , replication_strategy = [ <NUM_LIT:2> , <NUM_LIT:2> ] ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> bootstrap ( <NUM_LIT:5> , '<STR_LIT>' ) <EOL> wait_for_up ( cluster , <NUM_LIT:5> ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:6> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> decommission ( <NUM_LIT:3> ) <EOL> decommission ( <NUM_LIT:4> ) <EOL> wait_for_down ( cluster , <NUM_LIT:3> , wait = True ) <EOL> wait_for_down ( cluster , <NUM_LIT:4> , wait = True ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:0> ) <EOL> responses = set ( ) <EOL> for node in [ <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:5> ] : <EOL> responses . add ( self . coordinator_stats . get_query_count ( node ) ) <EOL> self . assertEqual ( set ( [ <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:12> ] ) , responses ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> decommission ( <NUM_LIT:5> ) <EOL> wait_for_down ( cluster , <NUM_LIT:5> , wait = True ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:0> ) <EOL> responses = set ( ) <EOL> for node in [ <NUM_LIT:1> , <NUM_LIT:2> ] : <EOL> responses . add ( self . coordinator_stats . get_query_count ( node ) ) <EOL> self . assertEqual ( set ( [ <NUM_LIT:0> , <NUM_LIT:12> ] ) , responses ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> decommission ( <NUM_LIT:1> ) <EOL> wait_for_down ( cluster , <NUM_LIT:1> , wait = True ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:12> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:4> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:5> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> force_stop ( <NUM_LIT:2> ) <EOL> try : <EOL> self . _query ( session , keyspace ) <EOL> self . fail ( ) <EOL> except NoHostAvailable : <EOL> pass <EOL> cluster . shutdown ( ) <EOL> def test_token_aware ( self ) : <EOL> keyspace = '<STR_LIT>' <EOL> self . token_aware ( keyspace ) <EOL> def test_token_aware_prepared ( self ) : <EOL> keyspace = '<STR_LIT>' <EOL> self . token_aware ( keyspace , True ) <EOL> def token_aware ( self , keyspace , use_prepared = False ) : <EOL> use_singledc ( ) <EOL> cluster = Cluster ( <EOL> load_balancing_policy = TokenAwarePolicy ( RoundRobinPolicy ( ) ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> ) <EOL> create_schema ( cluster , session , keyspace , replication_factor = <NUM_LIT:1> ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace , use_prepared = use_prepared ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:12> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> self . _query ( session , keyspace , use_prepared = use_prepared ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:12> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> force_stop ( <NUM_LIT:2> ) <EOL> wait_for_down ( cluster , <NUM_LIT:2> , wait = True ) <EOL> try : <EOL> self . _query ( session , keyspace , use_prepared = use_prepared ) <EOL> self . fail ( ) <EOL> except Unavailable as e : <EOL> self . assertEqual ( e . consistency , <NUM_LIT:1> ) <EOL> self . assertEqual ( e . required_replicas , <NUM_LIT:1> ) <EOL> self . assertEqual ( e . alive_replicas , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> start ( <NUM_LIT:2> ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = True ) <EOL> self . _query ( session , keyspace , use_prepared = use_prepared ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:12> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> stop ( <NUM_LIT:2> ) <EOL> wait_for_down ( cluster , <NUM_LIT:2> , wait = True ) <EOL> try : <EOL> self . _query ( session , keyspace , use_prepared = use_prepared ) <EOL> self . fail ( ) <EOL> except Unavailable : <EOL> pass <EOL> self . coordinator_stats . reset_counts ( ) <EOL> start ( <NUM_LIT:2> ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = True ) <EOL> decommission ( <NUM_LIT:2> ) <EOL> wait_for_down ( cluster , <NUM_LIT:2> , wait = True ) <EOL> self . _query ( session , keyspace , use_prepared = use_prepared ) <EOL> results = set ( [ <EOL> self . coordinator_stats . get_query_count ( <NUM_LIT:1> ) , <EOL> self . coordinator_stats . get_query_count ( <NUM_LIT:3> ) <EOL> ] ) <EOL> self . assertEqual ( results , set ( [ <NUM_LIT:0> , <NUM_LIT:12> ] ) ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> cluster . shutdown ( ) <EOL> def test_token_aware_composite_key ( self ) : <EOL> use_singledc ( ) <EOL> keyspace = '<STR_LIT>' <EOL> table = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = TokenAwarePolicy ( RoundRobinPolicy ( ) ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> ) <EOL> create_schema ( cluster , session , keyspace , replication_factor = <NUM_LIT:2> ) <EOL> session . execute ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % table ) <EOL> prepared = session . prepare ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' % table ) <EOL> session . execute ( prepared . bind ( ( <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> ) ) ) <EOL> results = session . execute ( '<STR_LIT>' % table ) <EOL> self . assertTrue ( results [ <NUM_LIT:0> ] . i ) <EOL> cluster . shutdown ( ) <EOL> def test_token_aware_with_rf_2 ( self , use_prepared = False ) : <EOL> use_singledc ( ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( <EOL> load_balancing_policy = TokenAwarePolicy ( RoundRobinPolicy ( ) ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> ) <EOL> create_schema ( cluster , session , keyspace , replication_factor = <NUM_LIT:2> ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:12> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> stop ( <NUM_LIT:2> ) <EOL> wait_for_down ( cluster , <NUM_LIT:2> , wait = True ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:12> ) <EOL> cluster . shutdown ( ) <EOL> def test_token_aware_with_local_table ( self ) : <EOL> use_singledc ( ) <EOL> cluster = Cluster ( <EOL> load_balancing_policy = TokenAwarePolicy ( RoundRobinPolicy ( ) ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> p = session . prepare ( "<STR_LIT>" ) <EOL> r = session . execute ( p , ( '<STR_LIT>' , ) ) <EOL> self . assertEqual ( r [ <NUM_LIT:0> ] . key , '<STR_LIT>' ) <EOL> cluster . shutdown ( ) <EOL> def test_white_list ( self ) : <EOL> use_singledc ( ) <EOL> keyspace = '<STR_LIT>' <EOL> cluster = Cluster ( ( '<STR_LIT>' , ) , <EOL> load_balancing_policy = WhiteListRoundRobinPolicy ( ( IP_FORMAT % <NUM_LIT:2> , ) ) , <EOL> protocol_version = PROTOCOL_VERSION ) <EOL> session = cluster . connect ( ) <EOL> wait_for_up ( cluster , <NUM_LIT:1> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:2> , wait = False ) <EOL> wait_for_up ( cluster , <NUM_LIT:3> ) <EOL> create_schema ( cluster , session , keyspace ) <EOL> self . _insert ( session , keyspace ) <EOL> self . _query ( session , keyspace ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:1> , <NUM_LIT:0> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:2> , <NUM_LIT:12> ) <EOL> self . coordinator_stats . assert_query_count_equals ( self , <NUM_LIT:3> , <NUM_LIT:0> ) <EOL> force_stop ( <NUM_LIT:3> ) <EOL> wait_for_down ( cluster , <NUM_LIT:3> ) <EOL> self . assertFalse ( cluster . metadata . _hosts [ IP_FORMAT % <NUM_LIT:3> ] . is_currently_reconnecting ( ) ) <EOL> self . coordinator_stats . reset_counts ( ) <EOL> force_stop ( <NUM_LIT:2> ) <EOL> time . sleep ( <NUM_LIT:10> ) <EOL> try : <EOL> self . _query ( session , keyspace ) <EOL> self . fail ( ) <EOL> except NoHostAvailable : <EOL> pass <EOL> cluster . shutdown ( ) </s>
