

def test_rfthree_tokenaware_none_down(self):
    keyspace = 'test_rfthree_tokenaware_none_down'
    cluster = Cluster(load_balancing_policy=TokenAwarePolicy(RoundRobinPolicy()), protocol_version=PROTOCOL_VERSION)
    session = cluster.connect()
    wait_for_up(cluster, 1, wait=False)
    wait_for_up(cluster, 2)
    create_schema(cluster, session, keyspace, replication_factor=3)
    self._insert(session, keyspace, count=1)
    self._query(session, keyspace, count=1)
    self.coordinator_stats.assert_query_count_equals(self, 1, 0)
    self.coordinator_stats.assert_query_count_equals(self, 2, 1)
    self.coordinator_stats.assert_query_count_equals(self, 3, 0)
    self.coordinator_stats.reset_counts()
    self._assert_writes_succeed(session, keyspace, SINGLE_DC_CONSISTENCY_LEVELS)
    self._assert_reads_succeed(session, keyspace, (SINGLE_DC_CONSISTENCY_LEVELS - set([ConsistencyLevel.ANY])), expected_reader=2)
    self.shutdown()

dataset/ETHPy150Open datastax/python-driver/tests/integration/long/test_consistency.py ConsistencyTests.test_rfthree_tokenaware_none_down/VarMisuse@181/274 `cluster`->`self`