def test_submit_schema_refresh(self):
        """
        Ensure new new schema is refreshed after submit_schema_refresh()
        """

        cluster = Cluster(protocol_version=PROTOCOL_VERSION)
        cluster.connect()
        self.assertNotIn("newkeyspace", cluster.metadata.keyspaces)

        other_cluster = Cluster(protocol_version=PROTOCOL_VERSION)
        session = other_cluster.connect()
        execute_until_pass(session,
            """
            CREATE KEYSPACE newkeyspace
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}
            """)

        future = cluster.submit_schema_refresh()
        future.result()

        self.assertIn("newkeyspace", cluster.metadata.keyspaces)

        execute_until_pass(session, "DROP KEYSPACE newkeyspace")
        cluster.shutdown()
        other_cluster.shutdown()

    def test_recreates(self):
        """
        Basic test for repeated schema creation and use, using many different keyspaces
        """

        session = self.session

        for i in range(2):
            for keyspace_number in range(5):
                keyspace = "ks_{0}".format(keyspace_number)

                if keyspace in self.cluster.metadata.keyspaces.keys():
                    drop = "DROP KEYSPACE {0}".format(keyspace)
                    log.debug(drop)
                    execute_until_pass(session, drop)

                create = "CREATE KEYSPACE {0} WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': 3}}".format(keyspace)
                log.debug(create)
                execute_until_pass(session, create)

                create = "CREATE TABLE {0}.cf (k int PRIMARY KEY, i int)".format(keyspace)
                log.debug(create)
                execute_until_pass(session, create)

                use = "USE {0}".format(keyspace)
                log.debug(use)
                execute_until_pass(session, use)

                insert = "INSERT INTO cf (k, i) VALUES (0, 0)"
                log.debug(insert)
                ss = SimpleStatement(insert, consistency_level=ConsistencyLevel.QUORUM)
                execute_until_pass(session, ss)

    def test_refresh_schema_type(self):
        if get_server_versions()[0] < (2, 1, 0):
            raise unittest.SkipTest('UDTs were introduced in Cassandra 2.1')

        if PROTOCOL_VERSION < 3:
            raise unittest.SkipTest('UDTs are not specified in change events for protocol v2')
            # We may want to refresh types on keyspace change events in that case(?)

        cluster = Cluster(protocol_version=PROTOCOL_VERSION)
        session = cluster.connect()

        keyspace_name = 'test1rf'
        type_name = self._testMethodName

        execute_until_pass(session, 'CREATE TYPE IF NOT EXISTS %s.%s (one int, two text)' % (keyspace_name, type_name))
        original_meta = cluster.metadata.keyspaces
        original_test1rf_meta = original_meta[keyspace_name]
        original_type_meta = original_test1rf_meta.user_types[type_name]

        # only refresh one type
        cluster.refresh_user_type_metadata('test1rf', type_name)
        current_meta = cluster.metadata.keyspaces
        current_test1rf_meta = current_meta[keyspace_name]
        current_type_meta = current_test1rf_meta.user_types[type_name]
        self.assertIs(original_meta, current_meta)
        self.assertEqual(original_test1rf_meta.export_as_string(), current_test1rf_meta.export_as_string())
        self.assertIsNot(original_type_meta, current_type_meta)
        self.assertEqual(original_type_meta.as_cql_query(), current_type_meta.as_cql_query())
        session.shutdown()

    def setUp(self):
        self._cass_version, self._cql_version = get_server_versions()

        if self._cass_version < (2, 1, 0):
            raise unittest.SkipTest("User Defined Types were introduced in Cassandra 2.1")

        self.cluster = Cluster(protocol_version=PROTOCOL_VERSION)
        self.session = self.cluster.connect()
        execute_until_pass(self.session,
                           "CREATE KEYSPACE udttests WITH replication = { 'class' : 'SimpleStrategy', 'replication_factor': '1'}")
        self.cluster.shutdown()

 def setUp(self):
     super(NameTupleFactory, self).setUp()
     self.session.row_factory = named_tuple_factory
     ddl = '''
             CREATE TABLE {0}.{1} (
                 k int PRIMARY KEY,
                 v1 text,
                 v2 text,
                 v3 text)'''.format(self.ks_name, self.function_table_name)
     self.session.execute(ddl)
     execute_until_pass(self.session, ddl)

    def test_can_insert_empty_values_for_int32(self):
        """
        Ensure Int32Type supports empty values
        """
        s = self.session

        execute_until_pass(s, "CREATE TABLE empty_values (a text PRIMARY KEY, b int)")
        execute_until_pass(s, "INSERT INTO empty_values (a, b) VALUES ('a', blobAsInt(0x))")
        try:
            Int32Type.support_empty_values = True
            results = execute_until_pass(s, "SELECT b FROM empty_values WHERE a='a'")[0]
            self.assertIs(EMPTY, results.b)
        finally:
            Int32Type.support_empty_values = False

    def test_read_timeout(self):
        """
        Trigger and ensure read_timeouts are counted
        Write a key, value pair. Pause a node without the coordinator node knowing about the "DOWN" state.
        Attempt a read at cl.ALL and receive a ReadTimeout.
        """


        # Test write
        self.session.execute("INSERT INTO test (k, v) VALUES (1, 1)")

        # Assert read
        query = SimpleStatement("SELECT * FROM test WHERE k=1", consistency_level=ConsistencyLevel.ALL)
        results = execute_until_pass(self.session, query)
        self.assertTrue(results)

        # Pause node so it shows as unreachable to coordinator
        get_node(1).pause()

        try:
            # Test read
            query = SimpleStatement("SELECT * FROM test", consistency_level=ConsistencyLevel.ALL)
            with self.assertRaises(ReadTimeout):
                self.session.execute(query, timeout=None)
            self.assertEqual(1, self.cluster.metrics.stats.read_timeouts)

        finally:
            get_node(1).resume()

    def test_write_timeout(self):
        """
        Trigger and ensure write_timeouts are counted
        Write a key, value pair. Pause a node without the coordinator node knowing about the "DOWN" state.
        Attempt a write at cl.ALL and receive a WriteTimeout.
        """

        cluster = Cluster(metrics_enabled=True, protocol_version=PROTOCOL_VERSION)
        session = cluster.connect("test3rf")

        # Test write
        session.execute("INSERT INTO test (k, v) VALUES (1, 1)")

        # Assert read
        query = SimpleStatement("SELECT * FROM test WHERE k=1", consistency_level=ConsistencyLevel.ALL)
        results = execute_until_pass(session, query)
        self.assertTrue(results)

        # Pause node so it shows as unreachable to coordinator
        get_node(1).pause()

        try:
            # Test write
            query = SimpleStatement("INSERT INTO test (k, v) VALUES (2, 2)", consistency_level=ConsistencyLevel.ALL)
            with self.assertRaises(WriteTimeout):
                session.execute(query, timeout=None)
            self.assertEqual(1, cluster.metrics.stats.write_timeouts)

        finally:
            get_node(1).resume()

        cluster.shutdown()

    def test_for_schema_disagreements_different_keyspaces(self):
        """
        Tests for any schema disagreements using many different keyspaces
        """

        session = self.session

        for i in range(30):
            execute_until_pass(session, "CREATE KEYSPACE test_{0} WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': 1}}".format(i))
            execute_until_pass(session, "CREATE TABLE test_{0}.cf (key int PRIMARY KEY, value int)".format(i))

            for j in range(100):
                execute_until_pass(session, "INSERT INTO test_{0}.cf (key, value) VALUES ({1}, {1})".format(i, j))

            execute_until_pass(session, "DROP KEYSPACE test_{0}".format(i))

    def test_basic(self):
        """
        Test basic connection and usage
        """

        cluster = Cluster(protocol_version=PROTOCOL_VERSION)
        session = cluster.connect()
        result = execute_until_pass(session,
            """
            CREATE KEYSPACE clustertests
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}
            """)
        self.assertEqual(None, result)

        result = execute_until_pass(session,
            """
            CREATE TABLE clustertests.cf0 (
                a text,
                b text,
                c text,
                PRIMARY KEY (a, b)
            )
            """)
        self.assertEqual(None, result)

        result = session.execute(
            """
            INSERT INTO clustertests.cf0 (a, b, c) VALUES ('a', 'b', 'c')
            """)
        self.assertEqual(None, result)

        result = session.execute("SELECT * FROM clustertests.cf0")
        self.assertEqual([('a', 'b', 'c')], result)

        execute_until_pass(session, "DROP KEYSPACE clustertests")

        cluster.shutdown()

    def _test_basic(self, dse_version):
        """
        Test basic connection and usage
        """
        cluster_name = '{}-{}'.format(
            self.__class__.__name__, dse_version.base_version.replace('.', '_')
        )
        use_cluster(cluster_name=cluster_name, nodes=[3],
                    dse_cluster=True, dse_options={}, dse_version=dse_version)

        cluster = Cluster(
            allow_beta_protocol_version=(dse_version >= Version('6.7.0')))
        session = cluster.connect()
        result = execute_until_pass(
            session,
            """
            CREATE KEYSPACE clustertests
            WITH replication = {'class': 'SimpleStrategy', 'replication_factor': '1'}
            """)
        self.assertFalse(result)

        result = execute_with_long_wait_retry(
            session,
            """
            CREATE TABLE clustertests.cf0 (
                a text,
                b text,
                c text,
                PRIMARY KEY (a, b)
            )
            """)
        self.assertFalse(result)

        result = session.execute(
            """
            INSERT INTO clustertests.cf0 (a, b, c) VALUES ('a', 'b', 'c')
            """)
        self.assertFalse(result)

        result = session.execute("SELECT * FROM clustertests.cf0")
        self.assertEqual([('a', 'b', 'c')], result)

        execute_with_long_wait_retry(session, "DROP KEYSPACE clustertests")

        cluster.shutdown()

    def test_unavailable(self):
        """
        Trigger and ensure unavailables are counted
        Write a key, value pair. Stop a node with the coordinator node knowing about the "DOWN" state.
        Attempt an insert/read at cl.ALL and receive a Unavailable Exception.
        """

        cluster = Cluster(metrics_enabled=True, protocol_version=PROTOCOL_VERSION)
        session = cluster.connect("test3rf")

        # Test write
        session.execute("INSERT INTO test (k, v) VALUES (1, 1)")

        # Assert read
        query = SimpleStatement("SELECT * FROM test WHERE k=1", consistency_level=ConsistencyLevel.ALL)
        results = execute_until_pass(session, query)
        self.assertTrue(results)

        # Stop node gracefully
        get_node(1).stop(wait=True, wait_other_notice=True)

        try:
            # Test write
            query = SimpleStatement("INSERT INTO test (k, v) VALUES (2, 2)", consistency_level=ConsistencyLevel.ALL)
            with self.assertRaises(Unavailable):
                session.execute(query)
            self.assertEqual(1, cluster.metrics.stats.unavailables)

            # Test write
            query = SimpleStatement("SELECT * FROM test", consistency_level=ConsistencyLevel.ALL)
            with self.assertRaises(Unavailable):
                session.execute(query, timeout=None)
            self.assertEqual(2, cluster.metrics.stats.unavailables)
        finally:
            get_node(1).start(wait_other_notice=True, wait_for_binary_proto=True)
            # Give some time for the cluster to come back up, for the next test
            time.sleep(5)

        cluster.shutdown()

    def nested_udt_schema_helper(self, session, MAX_NESTING_DEPTH):
        # create the seed udt
        execute_until_pass(session, "CREATE TYPE depth_0 (age int, name text)")

        # create the nested udts
        for i in range(MAX_NESTING_DEPTH):
            execute_until_pass(session, "CREATE TYPE depth_{0} (value frozen<depth_{1}>)".format(i + 1, i))

        # create a table with multiple sizes of nested udts
        # no need for all nested types, only a spot checked few and the largest one
        execute_until_pass(session, "CREATE TABLE mytable ("
                                    "k int PRIMARY KEY, "
                                    "v_0 frozen<depth_0>, "
                                    "v_1 frozen<depth_1>, "
                                    "v_2 frozen<depth_2>, "
                                    "v_3 frozen<depth_3>, "
                                    "v_{0} frozen<depth_{0}>)".format(MAX_NESTING_DEPTH))

    def test_unavailable(self):
        """
        Trigger and ensure unavailables are counted
        Write a key, value pair. Stop a node with the coordinator node knowing about the "DOWN" state.
        Attempt an insert/read at cl.ALL and receive a Unavailable Exception.
        """

        # Test write
        self.session.execute("INSERT INTO test (k, v) VALUES (1, 1)")

        # Assert read
        query = SimpleStatement("SELECT * FROM test WHERE k=1", consistency_level=ConsistencyLevel.ALL)
        results = execute_until_pass(self.session, query)
        self.assertTrue(results)

        # Stop node gracefully
        # Sometimes this commands continues with the other nodes having not noticed
        # 1 is down, and a Timeout error is returned instead of an Unavailable
        get_node(1).stop(wait=True, wait_other_notice=True)
        time.sleep(5)
        try:
            # Test write
            query = SimpleStatement("INSERT INTO test (k, v) VALUES (2, 2)", consistency_level=ConsistencyLevel.ALL)
            with self.assertRaises(Unavailable):
                self.session.execute(query)
            self.assertEqual(self.cluster.metrics.stats.unavailables, 1)

            # Test write
            query = SimpleStatement("SELECT * FROM test", consistency_level=ConsistencyLevel.ALL)
            with self.assertRaises(Unavailable):
                self.session.execute(query, timeout=None)
            self.assertEqual(self.cluster.metrics.stats.unavailables, 2)
        finally:
            get_node(1).start(wait_other_notice=True, wait_for_binary_proto=True)
            # Give some time for the cluster to come back up, for the next test
            time.sleep(5)

        self.cluster.shutdown()

 def tearDown(self):
     execute_until_pass(self.session, "DROP KEYSPACE udttests")
     self.cluster.shutdown()

 def _insert(self, session, keyspace, count, consistency_level=ConsistencyLevel.ONE):
     session.execute('USE %s' % keyspace)
     for i in range(count):
         ss = SimpleStatement('INSERT INTO cf(k, i) VALUES (0, 0)',
                              consistency_level=consistency_level)
         execute_until_pass(session, ss)

    def test_for_schema_disagreements_same_keyspace(self):
        """
        Tests for any schema disagreements using the same keyspace multiple times
        """

        cluster = Cluster(protocol_version=PROTOCOL_VERSION)
        session = cluster.connect(wait_for_all_pools=True)

        for i in range(30):
            try:
                execute_until_pass(session, "CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}")
            except AlreadyExists:
                execute_until_pass(session, "DROP KEYSPACE test")
                execute_until_pass(session, "CREATE KEYSPACE test WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}")

            execute_until_pass(session, "CREATE TABLE test.cf (key int PRIMARY KEY, value int)")

            for j in range(100):
                execute_until_pass(session, "INSERT INTO test.cf (key, value) VALUES ({0}, {0})".format(j))

            execute_until_pass(session, "DROP KEYSPACE test")

 def tearDown(self):
     self.cluster = Cluster(protocol_version=PROTOCOL_VERSION)
     self.session = self.cluster.connect()
     execute_until_pass(self.session, "DROP KEYSPACE udttests")
     self.cluster.shutdown()

 def tearDownClass(cls):
     execute_until_pass(cls.session, "DROP KEYSPACE typetests")
     cls.cluster.shutdown()

    def test_can_insert_empty_strings_and_nulls(self):
        """
        Test insertion of empty strings and null values
        """
        s = self.session

        # create table
        alpha_type_list = ["zz int PRIMARY KEY"]
        col_names = []
        string_types = set(('ascii', 'text', 'varchar'))
        string_columns = set((''))
        # this is just a list of types to try with empty strings
        non_string_types = PRIMITIVE_DATATYPES - string_types - set(('blob', 'date', 'inet', 'time', 'timestamp'))
        non_string_columns = set()
        start_index = ord('a')
        for i, datatype in enumerate(PRIMITIVE_DATATYPES):
            col_name = chr(start_index + i)
            alpha_type_list.append("{0} {1}".format(col_name, datatype))
            col_names.append(col_name)
            if datatype in non_string_types:
                non_string_columns.add(col_name)
            if datatype in string_types:
                string_columns.add(col_name)

        execute_until_pass(s, "CREATE TABLE all_empty ({0})".format(', '.join(alpha_type_list)))

        # verify all types initially null with simple statement
        columns_string = ','.join(col_names)
        s.execute("INSERT INTO all_empty (zz) VALUES (2)")
        results = s.execute("SELECT {0} FROM all_empty WHERE zz=2".format(columns_string))[0]
        self.assertTrue(all(x is None for x in results))

        # verify all types initially null with prepared statement
        select = s.prepare("SELECT {0} FROM all_empty WHERE zz=?".format(columns_string))
        results = s.execute(select.bind([2]))[0]
        self.assertTrue(all(x is None for x in results))

        # insert empty strings for string-like fields
        expected_values = dict((col, '') for col in string_columns)
        columns_string = ','.join(string_columns)
        placeholders = ','.join(["%s"] * len(string_columns))
        s.execute("INSERT INTO all_empty (zz, {0}) VALUES (3, {1})".format(columns_string, placeholders), expected_values.values())

        # verify string types empty with simple statement
        results = s.execute("SELECT {0} FROM all_empty WHERE zz=3".format(columns_string))[0]
        for expected, actual in zip(expected_values.values(), results):
            self.assertEqual(actual, expected)

        # verify string types empty with prepared statement
        results = s.execute(s.prepare("SELECT {0} FROM all_empty WHERE zz=?".format(columns_string)), [3])[0]
        for expected, actual in zip(expected_values.values(), results):
            self.assertEqual(actual, expected)

        # non-string types shouldn't accept empty strings
        for col in non_string_columns:
            query = "INSERT INTO all_empty (zz, {0}) VALUES (4, %s)".format(col)
            with self.assertRaises(InvalidRequest):
                s.execute(query, [''])

            insert = s.prepare("INSERT INTO all_empty (zz, {0}) VALUES (4, ?)".format(col))
            with self.assertRaises(TypeError):
                s.execute(insert, [''])

        # verify that Nones can be inserted and overwrites existing data
        # create the input
        params = []
        for datatype in PRIMITIVE_DATATYPES:
            params.append((get_sample(datatype)))

        # insert the data
        columns_string = ','.join(col_names)
        placeholders = ','.join(["%s"] * len(col_names))
        simple_insert = "INSERT INTO all_empty (zz, {0}) VALUES (5, {1})".format(columns_string, placeholders)
        s.execute(simple_insert, params)

        # then insert None, which should null them out
        null_values = [None] * len(col_names)
        s.execute(simple_insert, null_values)

        # check via simple statement
        query = "SELECT {0} FROM all_empty WHERE zz=5".format(columns_string)
        results = s.execute(query)[0]
        for col in results:
            self.assertEqual(None, col)

        # check via prepared statement
        select = s.prepare("SELECT {0} FROM all_empty WHERE zz=?".format(columns_string))
        results = s.execute(select.bind([5]))[0]
        for col in results:
            self.assertEqual(None, col)

        # do the same thing again, but use a prepared statement to insert the nulls
        s.execute(simple_insert, params)

        placeholders = ','.join(["?"] * len(col_names))
        insert = s.prepare("INSERT INTO all_empty (zz, {0}) VALUES (5, {1})".format(columns_string, placeholders))
        s.execute(insert, null_values)

        results = s.execute(query)[0]
        for col in results:
#.........这里部分代码省略.........