cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From aweisb...@apache.org
Subject [27/36] cassandra-dtest git commit: Migrate dtests to use pytest and python3
Date Mon, 29 Jan 2018 21:10:30 GMT
http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/dtest.py
----------------------------------------------------------------------
diff --git a/dtest.py b/dtest.py
index 25c52e3..aad9e58 100644
--- a/dtest.py
+++ b/dtest.py
@@ -1,50 +1,30 @@
-from __future__ import with_statement
-
-import ConfigParser
+import configparser
 import copy
-import errno
-import glob
 import logging
 import os
-import pprint
 import re
-import shutil
-import signal
 import subprocess
 import sys
-import tempfile
-import thread
 import threading
 import time
 import traceback
-import types
-import unittest.case
-from collections import OrderedDict
-from subprocess import CalledProcessError
-from unittest import TestCase
-
+import pytest
 import cassandra
 import ccmlib.repository
-from cassandra import ConsistencyLevel
+
+from subprocess import CalledProcessError
+
+from flaky import flaky
+
+from cassandra import ConsistencyLevel, OperationTimedOut
 from cassandra.auth import PlainTextAuthProvider
-from cassandra.cluster import Cluster as PyCluster
-from cassandra.cluster import NoHostAvailable
-from cassandra.cluster import ExecutionProfile, EXEC_PROFILE_DEFAULT
-from cassandra.policies import RetryPolicy, WhiteListRoundRobinPolicy
-from ccmlib.cluster import Cluster
-from ccmlib.cluster_factory import ClusterFactory
+from cassandra.cluster import ExecutionProfile
+from cassandra.policies import RetryPolicy, RoundRobinPolicy
 from ccmlib.common import get_version_from_build, is_win
+from ccmlib.node import ToolError, TimeoutError
 from distutils.version import LooseVersion
-from nose.exc import SkipTest
-from nose.tools import assert_greater_equal
-from six import print_
+from tools.misc import retry_till_success
 
-from plugins.dtestconfig import _CONFIG as CONFIG
-# We don't want test files to know about the plugins module, so we import
-# constants here and re-export them.
-from plugins.dtestconfig import GlobalConfigObject
-from tools.context import log_filter
-from tools.funcutils import merge_dicts
 
 LOG_SAVED_DIR = "logs"
 try:
@@ -57,52 +37,14 @@ LAST_LOG = os.path.join(LOG_SAVED_DIR, "last")
 LAST_TEST_DIR = 'last_test_dir'
 
 DEFAULT_DIR = './'
-config = ConfigParser.RawConfigParser()
+config = configparser.RawConfigParser()
 if len(config.read(os.path.expanduser('~/.cassandra-dtest'))) > 0:
     if config.has_option('main', 'default_dir'):
         DEFAULT_DIR = os.path.expanduser(config.get('main', 'default_dir'))
-CASSANDRA_DIR = os.environ.get('CASSANDRA_DIR', DEFAULT_DIR)
-
-NO_SKIP = os.environ.get('SKIP', '').lower() in ('no', 'false')
-DEBUG = os.environ.get('DEBUG', '').lower() in ('yes', 'true')
-TRACE = os.environ.get('TRACE', '').lower() in ('yes', 'true')
-KEEP_LOGS = os.environ.get('KEEP_LOGS', '').lower() in ('yes', 'true')
-KEEP_TEST_DIR = os.environ.get('KEEP_TEST_DIR', '').lower() in ('yes', 'true')
-PRINT_DEBUG = os.environ.get('PRINT_DEBUG', '').lower() in ('yes', 'true')
-OFFHEAP_MEMTABLES = os.environ.get('OFFHEAP_MEMTABLES', '').lower() in ('yes', 'true')
-NUM_TOKENS = os.environ.get('NUM_TOKENS', '256')
-RECORD_COVERAGE = os.environ.get('RECORD_COVERAGE', '').lower() in ('yes', 'true')
-IGNORE_REQUIRE = os.environ.get('IGNORE_REQUIRE', '').lower() in ('yes', 'true')
-DATADIR_COUNT = os.environ.get('DATADIR_COUNT', '3')
-ENABLE_ACTIVE_LOG_WATCHING = os.environ.get('ENABLE_ACTIVE_LOG_WATCHING', '').lower() in ('yes', 'true')
-RUN_STATIC_UPGRADE_MATRIX = os.environ.get('RUN_STATIC_UPGRADE_MATRIX', '').lower() in ('yes', 'true')
-
-# devault values for configuration from configuration plugin
-_default_config = GlobalConfigObject(
-    vnodes=True,
-)
-
-if CONFIG is None:
-    CONFIG = _default_config
 
-DISABLE_VNODES = not CONFIG.vnodes
-
-
-if os.environ.get('DISABLE_VNODES', '').lower() in ('yes', 'true'):
-    print 'DISABLE_VNODES environment variable deprecated. Use `./run_dtests.py --vnodes false` instead.'
-
-
-CURRENT_TEST = ""
-
-logging.basicConfig(filename=os.path.join(LOG_SAVED_DIR, "dtest.log"),
-                    filemode='w',
-                    format='%(asctime)s,%(msecs)d %(name)s %(levelname)s %(message)s',
-                    datefmt='%H:%M:%S',
-                    level=logging.DEBUG)
+RUN_STATIC_UPGRADE_MATRIX = os.environ.get('RUN_STATIC_UPGRADE_MATRIX', '').lower() in ('yes', 'true')
 
-LOG = logging.getLogger('dtest')
-# set python-driver log level to INFO by default for dtest
-logging.getLogger('cassandra').setLevel(logging.INFO)
+logger = logging.getLogger(__name__)
 
 
 def get_sha(repo_dir):
@@ -111,11 +53,13 @@ def get_sha(repo_dir):
         prefix = 'github:apache/'
         local_repo_location = os.environ.get('LOCAL_GIT_REPO')
         if local_repo_location is not None:
-            prefix = 'local:{}:'.format(local_repo_location)  # local: slugs take the form 'local:/some/path/to/cassandra/:branch_name_or_sha'
+            prefix = 'local:{}:'.format(local_repo_location)
+            # local: slugs take the form 'local:/some/path/to/cassandra/:branch_name_or_sha'
         return "{}{}".format(prefix, output)
     except CalledProcessError as e:
-        if re.search('Not a git repository', e.message) is not None:
-            # we tried to get a sha, but repo_dir isn't a git repo. No big deal, must just be working from a non-git install.
+        if re.search(str(e), 'Not a git repository') is not None:
+            # we tried to get a sha, but repo_dir isn't a git repo. No big deal, must just be
+            # working from a non-git install.
             return None
         else:
             # git call failed for some unknown reason
@@ -137,33 +81,12 @@ if _cassandra_version_slug:
     CASSANDRA_VERSION_FROM_BUILD = get_version_from_build(ccm_repo_cache_dir)
     CASSANDRA_GITREF = get_sha(ccm_repo_cache_dir)  # will be set None when not a git repo
 else:
-    CASSANDRA_VERSION_FROM_BUILD = get_version_from_build(CASSANDRA_DIR)
-    CASSANDRA_GITREF = get_sha(CASSANDRA_DIR)
-
+    CASSANDRA_VERSION_FROM_BUILD = LooseVersion("4.0") # todo kjkjkj
+    CASSANDRA_GITREF = ""
+    #CASSANDRA_VERSION_FROM_BUILD = get_version_from_build(self.dtest_config.cassandra_dir)
+    #CASSANDRA_GITREF = get_sha(dtest_config.cassandra_dir)
 
-# Determine the location of the libjemalloc jar so that we can specify it
-# through environment variables when start Cassandra.  This reduces startup
-# time, making the dtests run faster.
-def find_libjemalloc():
-    if is_win():
-        # let the normal bat script handle finding libjemalloc
-        return ""
 
-    this_dir = os.path.dirname(os.path.realpath(__file__))
-    script = os.path.join(this_dir, "findlibjemalloc.sh")
-    try:
-        p = subprocess.Popen([script], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
-        stdout, stderr = p.communicate()
-        if stderr or not stdout:
-            return "-"  # tells C* not to look for libjemalloc
-        else:
-            return stdout
-    except Exception as exc:
-        print "Failed to run script to prelocate libjemalloc ({}): {}".format(script, exc)
-        return ""
-
-
-CASSANDRA_LIBJEMALLOC = find_libjemalloc()
 # copy the initial environment variables so we can reset them later:
 initial_environment = copy.deepcopy(os.environ)
 
@@ -172,40 +95,7 @@ class DtestTimeoutError(Exception):
     pass
 
 
-def reset_environment_vars():
-    os.environ.clear()
-    os.environ.update(initial_environment)
-
-
-def warning(msg):
-    LOG.warning("{} - {}".format(CURRENT_TEST, msg))
-    if PRINT_DEBUG:
-        print "WARN: " + msg
-
-
-def debug(msg):
-    LOG.debug("{} - {}".format(CURRENT_TEST, msg))
-    if PRINT_DEBUG:
-        print msg
-
-
-debug("Python driver version in use: {}".format(cassandra.__version__))
-
-
-def retry_till_success(fun, *args, **kwargs):
-    timeout = kwargs.pop('timeout', 60)
-    bypassed_exception = kwargs.pop('bypassed_exception', Exception)
-
-    deadline = time.time() + timeout
-    while True:
-        try:
-            return fun(*args, **kwargs)
-        except bypassed_exception:
-            if time.time() > deadline:
-                raise
-            else:
-                # brief pause before next attempt
-                time.sleep(0.25)
+logger.debug("Python driver version in use: {}".format(cassandra.__version__))
 
 
 class FlakyRetryPolicy(RetryPolicy):
@@ -219,21 +109,21 @@ class FlakyRetryPolicy(RetryPolicy):
 
     def on_read_timeout(self, *args, **kwargs):
         if kwargs['retry_num'] < self.max_retries:
-            debug("Retrying read after timeout. Attempt #" + str(kwargs['retry_num']))
+            logger.debug("Retrying read after timeout. Attempt #" + str(kwargs['retry_num']))
             return (self.RETRY, None)
         else:
             return (self.RETHROW, None)
 
     def on_write_timeout(self, *args, **kwargs):
         if kwargs['retry_num'] < self.max_retries:
-            debug("Retrying write after timeout. Attempt #" + str(kwargs['retry_num']))
+            logger.debug("Retrying write after timeout. Attempt #" + str(kwargs['retry_num']))
             return (self.RETRY, None)
         else:
             return (self.RETHROW, None)
 
     def on_unavailable(self, *args, **kwargs):
         if kwargs['retry_num'] < self.max_retries:
-            debug("Retrying request after UE. Attempt #" + str(kwargs['retry_num']))
+            logger.debug("Retrying request after UE. Attempt #" + str(kwargs['retry_num']))
             return (self.RETRY, None)
         else:
             return (self.RETHROW, None)
@@ -261,8 +151,18 @@ class Runner(threading.Thread):
             i = i + 1
 
     def stop(self):
+        if self.__stopped:
+            return
+
         self.__stopped = True
-        self.join()
+        # pytests may appear to hang forever waiting for cluster tear down. are all driver session objects shutdown?
+        # to debug hang you can add the following at the top of the test
+        #     import faulthandler
+        #     faulthandler.enable()
+        #
+        # and then when the hang occurs send a SIGABRT to the pytest process (e.g. kill -SIGABRT <pytest_pid>)
+        # this will print a python thread dump of all currently alive threads
+        self.join(timeout=30)
         if self.__error is not None:
             raise self.__error
 
@@ -272,344 +172,103 @@ class Runner(threading.Thread):
 
 
 def make_execution_profile(retry_policy=FlakyRetryPolicy(), consistency_level=ConsistencyLevel.ONE, **kwargs):
-    return ExecutionProfile(retry_policy=retry_policy,
-                            consistency_level=consistency_level,
-                            **kwargs)
-
-
-class Tester(TestCase):
-
-    maxDiff = None
-    allow_log_errors = False  # scan the log of each node for errors after every test.
-    cluster_options = None
-
-    def set_node_to_current_version(self, node):
-        version = os.environ.get('CASSANDRA_VERSION')
-        cdir = CASSANDRA_DIR
-
-        if version:
-            node.set_install_dir(version=version)
-        else:
-            node.set_install_dir(install_dir=cdir)
-
-    def init_config(self):
-        init_default_config(self.cluster, self.cluster_options)
-
-    def setUp(self):
-        self.set_current_tst_name()
-        kill_windows_cassandra_procs()
-        maybe_cleanup_cluster_from_last_test_file()
-
-        self.test_path = get_test_path()
-        self.cluster = create_ccm_cluster(self.test_path, name='test')
-
-        self.maybe_begin_active_log_watch()
-        maybe_setup_jacoco(self.test_path)
-
-        self.init_config()
-        write_last_test_file(self.test_path, self.cluster)
-
-        set_log_levels(self.cluster)
-        self.connections = []
-        self.runners = []
-
-    # this is intentionally spelled 'tst' instead of 'test' to avoid
-    # making unittest think it's a test method
-    def set_current_tst_name(self):
-        global CURRENT_TEST
-        CURRENT_TEST = self.id()
-
-    def maybe_begin_active_log_watch(self):
-        if ENABLE_ACTIVE_LOG_WATCHING:
-            if not self.allow_log_errors:
-                self.begin_active_log_watch()
-
-    def begin_active_log_watch(self):
-        """
-        Calls into ccm to start actively watching logs.
-
-        In the event that errors are seen in logs, ccm will call back to _log_error_handler.
-
-        When the cluster is no longer in use, stop_active_log_watch should be called to end log watching.
-        (otherwise a 'daemon' thread will (needlessly) run until the process exits).
-        """
-        # log watching happens in another thread, but we want it to halt the main
-        # thread's execution, which we have to do by registering a signal handler
-        signal.signal(signal.SIGINT, self._catch_interrupt)
-        self._log_watch_thread = self.cluster.actively_watch_logs_for_error(self._log_error_handler, interval=0.25)
-
-    def _log_error_handler(self, errordata):
-        """
-        Callback handler used in conjunction with begin_active_log_watch.
-        When called, prepares exception instance, then will indirectly
-        cause _catch_interrupt to be called, which can raise the exception in the main
-        program thread.
-
-        @param errordata is a dictonary mapping node name to failure list.
-        """
-        # in some cases self.allow_log_errors may get set after proactive log checking has been enabled
-        # so we need to double-check first thing before proceeding
-        if self.allow_log_errors:
-            return
-
-        reportable_errordata = OrderedDict()
-
-        for nodename, errors in errordata.items():
-            filtered_errors = list(self.__filter_errors(['\n'.join(msg) for msg in errors]))
-            if len(filtered_errors) is not 0:
-                reportable_errordata[nodename] = filtered_errors
+    if 'load_balancing_policy' in kwargs:
+        return ExecutionProfile(retry_policy=retry_policy,
+                                consistency_level=consistency_level,
+                                **kwargs)
+    else:
+        return ExecutionProfile(retry_policy=retry_policy,
+                                consistency_level=consistency_level,
+                                load_balancing_policy=RoundRobinPolicy(),
+                                **kwargs)
 
-        # no errors worthy of halting the test
-        if not reportable_errordata:
-            return
 
-        message = "Errors seen in logs for: {nodes}".format(nodes=", ".join(reportable_errordata.keys()))
-        for nodename, errors in reportable_errordata.items():
-            for error in errors:
-                message += "\n{nodename}: {error}".format(nodename=nodename, error=error)
+def running_in_docker():
+    return os.path.isfile('/.dockerenv')
 
-        try:
-            debug('Errors were just seen in logs, ending test (if not ending already)!')
-            print_("Error details: \n{message}".format(message=message))
-            self.test_is_ending  # will raise AttributeError if not present
-        except AttributeError:
-            self.test_is_ending = True
-            self.exit_with_exception = AssertionError("Log error encountered during active log scanning, see stdout")
-            # thread.interrupt_main will SIGINT in the main thread, which we can
-            # catch to raise an exception with useful information
-            thread.interrupt_main()
 
+def cleanup_docker_environment_before_test_execution():
+    """
+    perform a bunch of system cleanup operations, like kill any instances that might be
+    hanging around incorrectly from a previous run, sync the disk, and clear swap.
+    Ideally we would also drop the page cache, but as docker isn't running in privileged
+    mode there is no way for us to do this.
     """
-    Finds files matching the glob pattern specified as argument on
-    the given keyspace in all nodes
+    # attempt to wack all existing running Cassandra processes forcefully to get us into a clean state
+    p_kill = subprocess.Popen('ps aux | grep -ie CassandraDaemon | grep java | awk \'{print $2}\' | xargs kill -9',
+                              shell=True)
+    p_kill.wait(timeout=10)
+
+    # explicitly call "sync" to flush everything that might be pending from a previous test
+    # so tests are less likely to hit a very slow fsync during the test by starting from a 'known' state
+    # note: to mitigate this further the docker image is mounting /tmp as a volume, which gives
+    # us an ext4 mount which should talk directly to the underlying device on the host, skipping
+    # the aufs pain that we get with anything else running in the docker image. Originally,
+    # I had a timeout of 120 seconds (2 minutes), 300 seconds (5 minutes) but sync was still occasionally timing out.
+    p_sync = subprocess.Popen('sudo /bin/sync', shell=True)
+    p_sync.wait(timeout=600)
+
+    # turn swap off and back on to make sure it's fully cleared if anything happened to swap
+    # from a previous test run
+    p_swap = subprocess.Popen('sudo /sbin/swapoff -a && sudo /sbin/swapon -a', shell=True)
+    p_swap.wait(timeout=60)
+
+
+def test_failure_due_to_timeout(err, *args):
     """
+    check if we should rerun a test with the flaky plugin or not.
+    for now, only run if we failed the test for one of the following
+    three exceptions: cassandra.OperationTimedOut, ccm.node.ToolError,
+    and ccm.node.TimeoutError.
+
+    - cassandra.OperationTimedOut will be thrown when a cql query made thru
+    the python-driver times out.
+    - ccm.node.ToolError will be thrown when an invocation of a "tool"
+    (in the case of dtests this will almost always invoking stress).
+    - ccm.node.TimeoutError will be thrown when a blocking ccm operation
+    on a individual node times out. In most cases this tends to be something
+    like watch_log_for hitting the timeout before the desired pattern is seen
+    in the node's logs.
+
+    if we failed for one of these reasons - and we're running in docker - run
+    the same "cleanup" logic we run before test execution and test setup begins
+    and for good measure introduce a 2 second sleep. why 2 seconds? because it's
+    magic :) - ideally this gets the environment back into a good state and makes
+    the rerun of flaky tests likely to suceed if they failed in the first place
+    due to environmental issues.
+    """
+    if issubclass(err[0], OperationTimedOut) or issubclass(err[0], ToolError) or issubclass(err[0], TimeoutError):
+        if running_in_docker():
+            cleanup_docker_environment_before_test_execution()
+            time.sleep(2)
+        return True
+    else:
+        return False
 
-    def glob_data_dirs(self, path, ks="ks"):
-        result = []
-        for node in self.cluster.nodelist():
-            for data_dir in node.data_directories():
-                ks_dir = os.path.join(data_dir, ks, path)
-                result.extend(glob.glob(ks_dir))
-        return result
-
-    def _catch_interrupt(self, signal, frame):
-        """
-        Signal handler for registering on SIGINT.
-
-        If called will look for a stored exception and raise it to abort test.
-        If a stored exception is not present, this handler has likely caught a
-        user interrupt via CTRL-C, and will raise a KeyboardInterrupt.
-        """
-        try:
-            # check if we have a persisted exception to fail with
-            raise self.exit_with_exception
-        except AttributeError:
-            # looks like this was just a plain CTRL-C event
-            raise KeyboardInterrupt()
-
-    def copy_logs(self, cluster, directory=None, name=None):
-        """Copy the current cluster's log files somewhere, by default to LOG_SAVED_DIR with a name of 'last'"""
-        if directory is None:
-            directory = LOG_SAVED_DIR
-        if name is None:
-            name = LAST_LOG
-        else:
-            name = os.path.join(directory, name)
-        if not os.path.exists(directory):
-            os.mkdir(directory)
-        logs = [(node.name, node.logfilename(), node.debuglogfilename(), node.gclogfilename(), node.compactionlogfilename())
-                for node in self.cluster.nodes.values()]
-        if len(logs) is not 0:
-            basedir = str(int(time.time() * 1000)) + '_' + self.id()
-            logdir = os.path.join(directory, basedir)
-            os.mkdir(logdir)
-            for n, log, debuglog, gclog, compactionlog in logs:
-                if os.path.exists(log):
-                    self.assertGreaterEqual(os.path.getsize(log), 0)
-                    shutil.copyfile(log, os.path.join(logdir, n + ".log"))
-                if os.path.exists(debuglog):
-                    self.assertGreaterEqual(os.path.getsize(debuglog), 0)
-                    shutil.copyfile(debuglog, os.path.join(logdir, n + "_debug.log"))
-                if os.path.exists(gclog):
-                    self.assertGreaterEqual(os.path.getsize(gclog), 0)
-                    shutil.copyfile(gclog, os.path.join(logdir, n + "_gc.log"))
-                if os.path.exists(compactionlog):
-                    self.assertGreaterEqual(os.path.getsize(compactionlog), 0)
-                    shutil.copyfile(compactionlog, os.path.join(logdir, n + "_compaction.log"))
-            if os.path.exists(name):
-                os.unlink(name)
-            if not is_win():
-                os.symlink(basedir, name)
-
-    def cql_connection(self, node, keyspace=None, user=None,
-                       password=None, compression=True, protocol_version=None, port=None, ssl_opts=None, **kwargs):
-
-        return self._create_session(node, keyspace, user, password, compression,
-                                    protocol_version, port=port, ssl_opts=ssl_opts, **kwargs)
-
-    def exclusive_cql_connection(self, node, keyspace=None, user=None,
-                                 password=None, compression=True, protocol_version=None, port=None, ssl_opts=None, **kwargs):
-
-        node_ip = get_ip_from_node(node)
-        wlrr = WhiteListRoundRobinPolicy([node_ip])
-
-        return self._create_session(node, keyspace, user, password, compression,
-                                    protocol_version, port=port, ssl_opts=ssl_opts, load_balancing_policy=wlrr, **kwargs)
-
-    def _create_session(self, node, keyspace, user, password, compression, protocol_version,
-                        port=None, ssl_opts=None, execution_profiles=None, **kwargs):
-        node_ip = get_ip_from_node(node)
-        if not port:
-            port = get_port_from_node(node)
-
-        if protocol_version is None:
-            protocol_version = get_eager_protocol_version(node.cluster.version())
-
-        if user is not None:
-            auth_provider = get_auth_provider(user=user, password=password)
-        else:
-            auth_provider = None
-
-        profiles = {EXEC_PROFILE_DEFAULT: make_execution_profile(**kwargs)
-                    } if not execution_profiles else execution_profiles
-
-        cluster = PyCluster([node_ip],
-                            auth_provider=auth_provider,
-                            compression=compression,
-                            protocol_version=protocol_version,
-                            port=port,
-                            ssl_options=ssl_opts,
-                            connect_timeout=10,
-                            allow_beta_protocol_version=True,
-                            execution_profiles=profiles)
-        session = cluster.connect(wait_for_all_pools=True)
-
-        if keyspace is not None:
-            session.set_keyspace(keyspace)
-
-        self.connections.append(session)
-        return session
-
-    def patient_cql_connection(self, node, keyspace=None,
-                               user=None, password=None, timeout=30, compression=True,
-                               protocol_version=None, port=None, ssl_opts=None, **kwargs):
-        """
-        Returns a connection after it stops throwing NoHostAvailables due to not being ready.
-
-        If the timeout is exceeded, the exception is raised.
-        """
-        if is_win():
-            timeout *= 2
-
-        expected_log_lines = ('Control connection failed to connect, shutting down Cluster:', '[control connection] Error connecting to ')
-        with log_filter('cassandra.cluster', expected_log_lines):
-            session = retry_till_success(
-                self.cql_connection,
-                node,
-                keyspace=keyspace,
-                user=user,
-                password=password,
-                timeout=timeout,
-                compression=compression,
-                protocol_version=protocol_version,
-                port=port,
-                ssl_opts=ssl_opts,
-                bypassed_exception=NoHostAvailable,
-                **kwargs
-            )
-
-        return session
-
-    def patient_exclusive_cql_connection(self, node, keyspace=None,
-                                         user=None, password=None, timeout=30, compression=True,
-                                         protocol_version=None, port=None, ssl_opts=None, **kwargs):
-        """
-        Returns a connection after it stops throwing NoHostAvailables due to not being ready.
-
-        If the timeout is exceeded, the exception is raised.
-        """
-        if is_win():
-            timeout *= 2
-
-        return retry_till_success(
-            self.exclusive_cql_connection,
-            node,
-            keyspace=keyspace,
-            user=user,
-            password=password,
-            timeout=timeout,
-            compression=compression,
-            protocol_version=protocol_version,
-            port=port,
-            ssl_opts=ssl_opts,
-            bypassed_exception=NoHostAvailable,
-            **kwargs
-        )
-
-    @classmethod
-    def tearDownClass(cls):
-        reset_environment_vars()
-        if os.path.exists(LAST_TEST_DIR):
-            with open(LAST_TEST_DIR) as f:
-                test_path = f.readline().strip('\n')
-                name = f.readline()
-                try:
-                    cluster = ClusterFactory.load(test_path, name)
-                    # Avoid waiting too long for node to be marked down
-                    if KEEP_TEST_DIR:
-                        cluster.stop(gently=RECORD_COVERAGE)
-                    else:
-                        cluster.remove()
-                        os.rmdir(test_path)
-                except IOError:
-                    # after a restart, /tmp will be emptied so we'll get an IOError when loading the old cluster here
-                    pass
-            try:
-                os.remove(LAST_TEST_DIR)
-            except IOError:
-                # Ignore - see comment above
-                pass
 
-    def tearDown(self):
-        # test_is_ending prevents active log watching from being able to interrupt the test
-        # which we don't want to happen once tearDown begins
-        self.test_is_ending = True
+@flaky(rerun_filter=test_failure_due_to_timeout)
+class Tester:
 
-        reset_environment_vars()
+    def __getattribute__(self, name):
+        try:
+            return object.__getattribute__(self, name)
+        except AttributeError:
+            fixture_dtest_setup = object.__getattribute__(self, 'fixture_dtest_setup')
+            return object.__getattribute__(fixture_dtest_setup , name)
 
-        for con in self.connections:
-            con.cluster.shutdown()
+    @pytest.fixture(scope='function', autouse=True)
+    def set_dtest_setup_on_function(self, fixture_dtest_setup, fixture_dtest_config):
+        self.fixture_dtest_setup = fixture_dtest_setup
+        self.dtest_config = fixture_dtest_config
 
-        for runner in self.runners:
-            try:
-                runner.stop()
-            except Exception:
-                pass
+    def set_node_to_current_version(self, node):
+        version = os.environ.get('CASSANDRA_VERSION')
 
-        failed = did_fail()
-        try:
-            if not self.allow_log_errors and self.check_logs_for_errors():
-                failed = True
-                raise AssertionError('Unexpected error in log, see stdout')
-        finally:
-            try:
-                # save the logs for inspection
-                if failed or KEEP_LOGS:
-                    self.copy_logs(self.cluster)
-            except Exception as e:
-                print "Error saving log:", str(e)
-            finally:
-                log_watch_thread = getattr(self, '_log_watch_thread', None)
-                cleanup_cluster(self.cluster, self.test_path, log_watch_thread)
-
-    def check_logs_for_errors(self):
-        for node in self.cluster.nodelist():
-            errors = list(self.__filter_errors(
-                ['\n'.join(msg) for msg in node.grep_log_for_errors()]))
-            if len(errors) is not 0:
-                for error in errors:
-                    print_("Unexpected error in {node_name} log, error: \n{error}".format(node_name=node.name, error=error))
-                return True
+        if version:
+            node.set_install_dir(version=version)
+        else:
+            node.set_install_dir(install_dir=self.dtest_config.cassandra_dir)
+            os.environ.set('CASSANDRA_DIR', self.dtest_config.cassandra_dir)
 
     def go(self, func):
         runner = Runner(func)
@@ -617,57 +276,6 @@ class Tester(TestCase):
         runner.start()
         return runner
 
-    def skip(self, msg):
-        if not NO_SKIP:
-            raise SkipTest(msg)
-
-    def __filter_errors(self, errors):
-        """Filter errors, removing those that match self.ignore_log_patterns"""
-        if not hasattr(self, 'ignore_log_patterns'):
-            self.ignore_log_patterns = []
-        for e in errors:
-            for pattern in self.ignore_log_patterns:
-                if re.search(pattern, e):
-                    break
-            else:
-                yield e
-
-    # Disable docstrings printing in nosetest output
-    def shortDescription(self):
-        return None
-
-    def get_jfr_jvm_args(self):
-        """
-        @return The JVM arguments required for attaching flight recorder to a Java process.
-        """
-        return ["-XX:+UnlockCommercialFeatures", "-XX:+FlightRecorder"]
-
-    def start_jfr_recording(self, nodes):
-        """
-        Start Java flight recorder provided the cluster was started with the correct jvm arguments.
-        """
-        for node in nodes:
-            p = subprocess.Popen(['jcmd', str(node.pid), 'JFR.start'],
-                                 stdout=subprocess.PIPE,
-                                 stderr=subprocess.PIPE)
-            stdout, stderr = p.communicate()
-            debug(stdout)
-            debug(stderr)
-
-    def dump_jfr_recording(self, nodes):
-        """
-        Save Java flight recorder results to file for analyzing with mission control.
-        """
-        for node in nodes:
-            p = subprocess.Popen(['jcmd', str(node.pid), 'JFR.dump',
-                                  'recording=1', 'filename=recording_{}.jfr'.format(node.address())],
-                                 stdout=subprocess.PIPE,
-                                 stderr=subprocess.PIPE)
-            stdout, stderr = p.communicate()
-            debug(stdout)
-            debug(stderr)
-
-
 def get_eager_protocol_version(cassandra_version):
     """
     Returns the highest protocol version accepted
@@ -690,7 +298,7 @@ def create_cf(session, name, key_type="varchar", speculative_retry=None, read_re
 
     additional_columns = ""
     if columns is not None:
-        for k, v in columns.items():
+        for k, v in list(columns.items()):
             additional_columns = "{}, {} {}".format(additional_columns, k, v)
 
     if additional_columns == "":
@@ -714,20 +322,42 @@ def create_cf(session, name, key_type="varchar", speculative_retry=None, read_re
     if compact_storage:
         query += ' AND COMPACT STORAGE'
 
-    session.execute(query)
-    time.sleep(0.2)
-
+    try:
+        retry_till_success(session.execute, query=query, timeout=120, bypassed_exception=cassandra.OperationTimedOut)
+    except cassandra.AlreadyExists:
+        logger.warn('AlreadyExists executing create cf query \'%s\'' % query)
+    session.cluster.control_connection.wait_for_schema_agreement(wait_time=120)
+    #Going to ignore OperationTimedOut from create CF, so need to validate it was indeed created
+    session.execute('SELECT * FROM %s LIMIT 1' % name);
+
+def create_cf_simple(session, name, query):
+    try:
+        retry_till_success(session.execute, query=query, timeout=120, bypassed_exception=cassandra.OperationTimedOut)
+    except cassandra.AlreadyExists:
+        logger.warn('AlreadyExists executing create cf query \'%s\'' % query)
+    session.cluster.control_connection.wait_for_schema_agreement(wait_time=120)
+    #Going to ignore OperationTimedOut from create CF, so need to validate it was indeed created
+    session.execute('SELECT * FROM %s LIMIT 1' % name)
 
 def create_ks(session, name, rf):
     query = 'CREATE KEYSPACE %s WITH replication={%s}'
-    if isinstance(rf, types.IntType):
+    if isinstance(rf, int):
         # we assume simpleStrategy
-        session.execute(query % (name, "'class':'SimpleStrategy', 'replication_factor':%d" % rf))
+        query = query % (name, "'class':'SimpleStrategy', 'replication_factor':%d" % rf)
     else:
-        assert_greater_equal(len(rf), 0, "At least one datacenter/rf pair is needed")
+        assert len(rf) >= 0, "At least one datacenter/rf pair is needed"
         # we assume networkTopologyStrategy
-        options = (', ').join(['\'%s\':%d' % (d, r) for d, r in rf.iteritems()])
-        session.execute(query % (name, "'class':'NetworkTopologyStrategy', %s" % options))
+        options = (', ').join(['\'%s\':%d' % (d, r) for d, r in rf.items()])
+        query = query % (name, "'class':'NetworkTopologyStrategy', %s" % options)
+
+    try:
+        retry_till_success(session.execute, query=query, timeout=120, bypassed_exception=cassandra.OperationTimedOut)
+    except cassandra.AlreadyExists:
+        logger.warn('AlreadyExists executing create ks query \'%s\'' % query)
+
+    session.cluster.control_connection.wait_for_schema_agreement(wait_time=120)
+    #Also validates it was indeed created even though we ignored OperationTimedOut
+    #Might happen some of the time because CircleCI disk IO is unreliable and hangs randomly
     session.execute('USE {}'.format(name))
 
 
@@ -774,301 +404,13 @@ def kill_windows_cassandra_procs():
                     pass
                 else:
                     if (pinfo['name'] == 'java.exe' and '-Dcassandra' in pinfo['cmdline']):
-                        print 'Found running cassandra process with pid: ' + str(pinfo['pid']) + '. Killing.'
+                        print('Found running cassandra process with pid: ' + str(pinfo['pid']) + '. Killing.')
                         psutil.Process(pinfo['pid']).kill()
         except ImportError:
-            debug("WARN: psutil not installed. Cannot detect and kill "
+            logger.debug("WARN: psutil not installed. Cannot detect and kill "
                   "running cassandra processes - you may see cascading dtest failures.")
 
 
-def get_test_path():
-    test_path = tempfile.mkdtemp(prefix='dtest-')
-
-    # ccm on cygwin needs absolute path to directory - it crosses from cygwin space into
-    # regular Windows space on wmic calls which will otherwise break pathing
-    if sys.platform == "cygwin":
-        process = subprocess.Popen(["cygpath", "-m", test_path], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-        test_path = process.communicate()[0].rstrip()
-
-    return test_path
-
-
-# nose will discover this as a test, so we manually make it not a test
-get_test_path.__test__ = False
-
-
-def create_ccm_cluster(test_path, name):
-    debug("cluster ccm directory: " + test_path)
-    version = os.environ.get('CASSANDRA_VERSION')
-    cdir = CASSANDRA_DIR
-
-    if version:
-        cluster = Cluster(test_path, name, cassandra_version=version)
-    else:
-        cluster = Cluster(test_path, name, cassandra_dir=cdir)
-
-    if DISABLE_VNODES:
-        cluster.set_configuration_options(values={'num_tokens': None})
-    else:
-        cluster.set_configuration_options(values={'initial_token': None, 'num_tokens': NUM_TOKENS})
-
-    if OFFHEAP_MEMTABLES:
-        cluster.set_configuration_options(values={'memtable_allocation_type': 'offheap_objects'})
-
-    cluster.set_datadir_count(DATADIR_COUNT)
-    cluster.set_environment_variable('CASSANDRA_LIBJEMALLOC', CASSANDRA_LIBJEMALLOC)
-
-    return cluster
-
-
-def cleanup_cluster(cluster, test_path, log_watch_thread=None):
-    with log_filter('cassandra'):  # quiet noise from driver when nodes start going down
-        if KEEP_TEST_DIR:
-            cluster.stop(gently=RECORD_COVERAGE)
-        else:
-            # when recording coverage the jvm has to exit normally
-            # or the coverage information is not written by the jacoco agent
-            # otherwise we can just kill the process
-            if RECORD_COVERAGE:
-                cluster.stop(gently=True)
-
-            # Cleanup everything:
-            try:
-                if log_watch_thread:
-                    stop_active_log_watch(log_watch_thread)
-            finally:
-                debug("removing ccm cluster {name} at: {path}".format(name=cluster.name, path=test_path))
-                cluster.remove()
-
-                debug("clearing ssl stores from [{0}] directory".format(test_path))
-                for filename in ('keystore.jks', 'truststore.jks', 'ccm_node.cer'):
-                    try:
-                        os.remove(os.path.join(test_path, filename))
-                    except OSError as e:
-                        # once we port to py3, which has better reporting for exceptions raised while
-                        # handling other excpetions, we should just assert e.errno == errno.ENOENT
-                        if e.errno != errno.ENOENT:  # ENOENT = no such file or directory
-                            raise
-
-                os.rmdir(test_path)
-                cleanup_last_test_dir()
-
-
-def cleanup_last_test_dir():
-    if os.path.exists(LAST_TEST_DIR):
-        os.remove(LAST_TEST_DIR)
-
-
-def stop_active_log_watch(log_watch_thread):
-    """
-    Joins the log watching thread, which will then exit.
-    Should be called after each test, ideally after nodes are stopped but before cluster files are removed.
-
-    Can be called multiple times without error.
-    If not called, log watching thread will remain running until the parent process exits.
-    """
-    log_watch_thread.join(timeout=60)
-
-
-def maybe_cleanup_cluster_from_last_test_file():
-    # cleaning up if a previous execution didn't trigger tearDown (which
-    # can happen if it is interrupted by KeyboardInterrupt)
-    if os.path.exists(LAST_TEST_DIR):
-        with open(LAST_TEST_DIR) as f:
-            test_path = f.readline().strip('\n')
-            name = f.readline()
-        try:
-            cluster = ClusterFactory.load(test_path, name)
-            # Avoid waiting too long for node to be marked down
-            cleanup_cluster(cluster, test_path)
-        except IOError:
-            # after a restart, /tmp will be emptied so we'll get an IOError when loading the old cluster here
-            pass
-
-
-def init_default_config(cluster, cluster_options):
-    # the failure detector can be quite slow in such tests with quick start/stop
-    phi_values = {'phi_convict_threshold': 5}
-
-    timeout = 10000
-    if cluster_options is not None:
-        values = merge_dicts(cluster_options, phi_values)
-    else:
-        values = merge_dicts(phi_values, {
-            'read_request_timeout_in_ms': timeout,
-            'range_request_timeout_in_ms': timeout,
-            'write_request_timeout_in_ms': timeout,
-            'truncate_request_timeout_in_ms': timeout,
-            'request_timeout_in_ms': timeout
-        })
-
-    # No more thrift in 4.0, and start_rpc doesn't exists anymore
-    if cluster.version() >= '4' and 'start_rpc' in values:
-        del values['start_rpc']
-
-    cluster.set_configuration_options(values)
-    debug("Done setting configuration options:\n" + pprint.pformat(cluster._config_options, indent=4))
-
-
-def write_last_test_file(test_path, cluster):
-    with open(LAST_TEST_DIR, 'w') as f:
-        f.write(test_path + '\n')
-        f.write(cluster.name)
-
-
-def set_log_levels(cluster):
-    if DEBUG:
-        cluster.set_log_level("DEBUG")
-    if TRACE:
-        cluster.set_log_level("TRACE")
-
-    if os.environ.get('DEBUG', 'no').lower() not in ('no', 'false', 'yes', 'true'):
-        classes_to_debug = os.environ.get('DEBUG').split(":")
-        cluster.set_log_level('DEBUG', None if len(classes_to_debug) == 0 else classes_to_debug)
-
-    if os.environ.get('TRACE', 'no').lower() not in ('no', 'false', 'yes', 'true'):
-        classes_to_trace = os.environ.get('TRACE').split(":")
-        cluster.set_log_level('TRACE', None if len(classes_to_trace) == 0 else classes_to_trace)
-
-
-def maybe_setup_jacoco(test_path, cluster_name='test'):
-    """Setup JaCoCo code coverage support"""
-
-    if not RECORD_COVERAGE:
-        return
-
-    # use explicit agent and execfile locations
-    # or look for a cassandra build if they are not specified
-    cdir = CASSANDRA_DIR
-
-    agent_location = os.environ.get('JACOCO_AGENT_JAR', os.path.join(cdir, 'build/lib/jars/jacocoagent.jar'))
-    jacoco_execfile = os.environ.get('JACOCO_EXECFILE', os.path.join(cdir, 'build/jacoco/jacoco.exec'))
-
-    if os.path.isfile(agent_location):
-        debug("Jacoco agent found at {}".format(agent_location))
-        with open(os.path.join(
-                test_path, cluster_name, 'cassandra.in.sh'), 'w') as f:
-
-            f.write('JVM_OPTS="$JVM_OPTS -javaagent:{jar_path}=destfile={exec_file}"'
-                    .format(jar_path=agent_location, exec_file=jacoco_execfile))
-
-            if os.path.isfile(jacoco_execfile):
-                debug("Jacoco execfile found at {}, execution data will be appended".format(jacoco_execfile))
-            else:
-                debug("Jacoco execfile will be created at {}".format(jacoco_execfile))
-    else:
-        debug("Jacoco agent not found or is not file. Execution will not be recorded.")
-
-
-def did_fail():
-    if sys.exc_info() == (None, None, None):
-        return False
-
-    exc_class, _, _ = sys.exc_info()
-    return not issubclass(exc_class, unittest.case.SkipTest)
-
-
-class ReusableClusterTester(Tester):
-    """
-    A Tester designed for reusing the same cluster across multiple
-    test methods.  This makes test suites with many small tests run
-    much, much faster.  However, there are a couple of downsides:
-
-    First, test setup and teardown must be diligent about cleaning
-    up any data or schema elements that may interfere with other
-    tests.
-
-    Second, errors triggered by one test method may cascade
-    into other test failures.  In an attempt to limit this, the
-    cluster will be restarted if a test fails or an exception is
-    caught.  However, there may still be undetected problems in
-    Cassandra that cause cascading failures.
-    """
-
-    test_path = None
-    cluster = None
-    cluster_options = None
-
-    @classmethod
-    def setUpClass(cls):
-        kill_windows_cassandra_procs()
-        maybe_cleanup_cluster_from_last_test_file()
-        cls.initialize_cluster()
-
-    def setUp(self):
-        self.set_current_tst_name()
-        self.connections = []
-
-        # TODO enable active log watching
-        # This needs to happen in setUp() and not setUpClass() so that individual
-        # test methods can set allow_log_errors and so that error handling
-        # only fails a single test method instead of the entire class.
-        # The problem with this is that ccm doesn't yet support stopping the
-        # active log watcher -- it runs until the cluster is destroyed.  Since
-        # we reuse the same cluster, this doesn't work for us.
-
-    def tearDown(self):
-        # test_is_ending prevents active log watching from being able to interrupt the test
-        self.test_is_ending = True
-
-        failed = did_fail()
-        try:
-            if not self.allow_log_errors and self.check_logs_for_errors():
-                failed = True
-                raise AssertionError('Unexpected error in log, see stdout')
-        finally:
-            try:
-                # save the logs for inspection
-                if failed or KEEP_LOGS:
-                    self.copy_logs(self.cluster)
-            except Exception as e:
-                print "Error saving log:", str(e)
-            finally:
-                reset_environment_vars()
-                if failed:
-                    cleanup_cluster(self.cluster, self.test_path)
-                    kill_windows_cassandra_procs()
-                    self.initialize_cluster()
-
-    @classmethod
-    def initialize_cluster(cls):
-        """
-        This method is responsible for initializing and configuring a ccm
-        cluster for the next set of tests.  This can be called for two
-        different reasons:
-         * A class of tests is starting
-         * A test method failed/errored, so the cluster has been wiped
-
-        Subclasses that require custom initialization should generally
-        do so by overriding post_initialize_cluster().
-        """
-        cls.test_path = get_test_path()
-        cls.cluster = create_ccm_cluster(cls.test_path, name='test')
-        cls.init_config()
-
-        maybe_setup_jacoco(cls.test_path)
-        cls.init_config()
-        write_last_test_file(cls.test_path, cls.cluster)
-        set_log_levels(cls.cluster)
-
-        cls.post_initialize_cluster()
-
-    @classmethod
-    def post_initialize_cluster(cls):
-        """
-        This method is called after the ccm cluster has been created
-        and default config options have been applied.  Any custom
-        initialization for a test class should generally be done
-        here in order to correctly handle cluster restarts after
-        test method failures.
-        """
-        pass
-
-    @classmethod
-    def init_config(cls):
-        init_default_config(cls.cluster, cls.cluster_options)
-
-
 class MultiError(Exception):
     """
     Extends Exception to provide reporting multiple exceptions at once.
@@ -1109,24 +451,20 @@ def run_scenarios(scenarios, handler, deferred_exceptions=tuple()):
     tracebacks = []
 
     for i, scenario in enumerate(scenarios, 1):
-        debug("running scenario {}/{}: {}".format(i, len(scenarios), scenario))
+        logger.debug("running scenario {}/{}: {}".format(i, len(scenarios), scenario))
 
         try:
             handler(scenario)
         except deferred_exceptions as e:
             tracebacks.append(traceback.format_exc(sys.exc_info()))
-            errors.append(type(e)('encountered {} {} running scenario:\n  {}\n'.format(e.__class__.__name__, e.message, scenario)))
-            debug("scenario {}/{} encountered a deferrable exception, continuing".format(i, len(scenarios)))
+            errors.append(type(e)('encountered {} {} running scenario:\n  {}\n'.format(e.__class__.__name__, str(e), scenario)))
+            logger.debug("scenario {}/{} encountered a deferrable exception, continuing".format(i, len(scenarios)))
         except Exception as e:
             # catch-all for any exceptions not intended to be deferred
             tracebacks.append(traceback.format_exc(sys.exc_info()))
-            errors.append(type(e)('encountered {} {} running scenario:\n  {}\n'.format(e.__class__.__name__, e.message, scenario)))
-            debug("scenario {}/{} encountered a non-deferrable exception, aborting".format(i, len(scenarios)))
+            errors.append(type(e)('encountered {} {} running scenario:\n  {}\n'.format(e.__class__.__name__, str(e), scenario)))
+            logger.debug("scenario {}/{} encountered a non-deferrable exception, aborting".format(i, len(scenarios)))
             raise MultiError(errors, tracebacks)
 
     if errors:
         raise MultiError(errors, tracebacks)
-
-
-def supports_v5_protocol(cluster_version):
-    return cluster_version >= LooseVersion('4.0')

http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/dtest_setup.py
----------------------------------------------------------------------
diff --git a/dtest_setup.py b/dtest_setup.py
new file mode 100644
index 0000000..87014f4
--- /dev/null
+++ b/dtest_setup.py
@@ -0,0 +1,498 @@
+import pytest
+import glob
+import os
+import shutil
+import time
+import logging
+import re
+import tempfile
+import subprocess
+import sys
+import errno
+import pprint
+from collections import OrderedDict
+
+from cassandra.cluster import Cluster as PyCluster
+from cassandra.cluster import NoHostAvailable
+from cassandra.cluster import EXEC_PROFILE_DEFAULT
+from cassandra.policies import WhiteListRoundRobinPolicy
+from ccmlib.common import get_version_from_build, is_win
+from ccmlib.cluster import Cluster
+
+from dtest import (get_ip_from_node, make_execution_profile, get_auth_provider, get_port_from_node,
+                   get_eager_protocol_version)
+from distutils.version import LooseVersion
+
+from tools.context import log_filter
+from tools.funcutils import merge_dicts
+
+logger = logging.getLogger(__name__)
+
+
+def retry_till_success(fun, *args, **kwargs):
+    timeout = kwargs.pop('timeout', 60)
+    bypassed_exception = kwargs.pop('bypassed_exception', Exception)
+
+    deadline = time.time() + timeout
+    while True:
+        try:
+            return fun(*args, **kwargs)
+        except bypassed_exception:
+            if time.time() > deadline:
+                raise
+            else:
+                # brief pause before next attempt
+                time.sleep(0.25)
+
+
+class DTestSetup:
+    def __init__(self, dtest_config=None, setup_overrides=None):
+        self.dtest_config = dtest_config
+        self.setup_overrides = setup_overrides
+        self.ignore_log_patterns = []
+        self.cluster = None
+        self.cluster_options = []
+        self.replacement_node = None
+        self.allow_log_errors = False
+        self.connections = []
+
+        self.log_saved_dir = "logs"
+        try:
+            os.mkdir(self.log_saved_dir)
+        except OSError:
+            pass
+
+        self.last_log = os.path.join(self.log_saved_dir, "last")
+        self.test_path = self.get_test_path()
+        self.enable_for_jolokia = False
+        self.subprocs = []
+        self.log_watch_thread = None
+        self.last_test_dir = "last_test_dir"
+        self.jvm_args = []
+
+    def get_test_path(self):
+        test_path = tempfile.mkdtemp(prefix='dtest-')
+
+        # ccm on cygwin needs absolute path to directory - it crosses from cygwin space into
+        # regular Windows space on wmic calls which will otherwise break pathing
+        if sys.platform == "cygwin":
+            process = subprocess.Popen(["cygpath", "-m", test_path], stdout=subprocess.PIPE,
+                                       stderr=subprocess.STDOUT)
+            test_path = process.communicate()[0].rstrip()
+
+        return test_path
+
+    def glob_data_dirs(self, path, ks="ks"):
+        result = []
+        for node in self.cluster.nodelist():
+            for data_dir in node.data_directories():
+                ks_dir = os.path.join(data_dir, ks, path)
+                result.extend(glob.glob(ks_dir))
+        return result
+
+    def begin_active_log_watch(self):
+        """
+        Calls into ccm to start actively watching logs.
+
+        In the event that errors are seen in logs, ccm will call back to _log_error_handler.
+
+        When the cluster is no longer in use, stop_active_log_watch should be called to end log watching.
+        (otherwise a 'daemon' thread will (needlessly) run until the process exits).
+        """
+        self._log_watch_thread = self.cluster.actively_watch_logs_for_error(self._log_error_handler, interval=0.25)
+
+    def _log_error_handler(self, errordata):
+        """
+        Callback handler used in conjunction with begin_active_log_watch.
+        When called, prepares exception instance, we will use pytest.fail
+        to kill the current test being executed and mark it as failed
+
+        @param errordata is a dictonary mapping node name to failure list.
+        """
+        # in some cases self.allow_log_errors may get set after proactive log checking has been enabled
+        # so we need to double-check first thing before proceeding
+        if self.allow_log_errors:
+            return
+
+        reportable_errordata = OrderedDict()
+
+        for nodename, errors in list(errordata.items()):
+            filtered_errors = list(self.__filter_errors(['\n'.join(msg) for msg in errors]))
+            if len(filtered_errors) is not 0:
+                reportable_errordata[nodename] = filtered_errors
+
+        # no errors worthy of halting the test
+        if not reportable_errordata:
+            return
+
+        message = "Errors seen in logs for: {nodes}".format(nodes=", ".join(list(reportable_errordata.keys())))
+        for nodename, errors in list(reportable_errordata.items()):
+            for error in errors:
+                message += "\n{nodename}: {error}".format(nodename=nodename, error=error)
+
+        logger.debug('Errors were just seen in logs, ending test (if not ending already)!')
+        pytest.fail("Error details: \n{message}".format(message=message))
+
+    def copy_logs(self, directory=None, name=None):
+        """Copy the current cluster's log files somewhere, by default to LOG_SAVED_DIR with a name of 'last'"""
+        if directory is None:
+            directory = self.log_saved_dir
+        if name is None:
+            name = self.last_log
+        else:
+            name = os.path.join(directory, name)
+        if not os.path.exists(directory):
+            os.mkdir(directory)
+        logs = [(node.name, node.logfilename(), node.debuglogfilename(), node.gclogfilename(),
+                 node.compactionlogfilename())
+                for node in list(self.cluster.nodes.values())]
+        if len(logs) is not 0:
+            basedir = str(int(time.time() * 1000)) + '_' + str(id(self))
+            logdir = os.path.join(directory, basedir)
+            os.mkdir(logdir)
+            for n, log, debuglog, gclog, compactionlog in logs:
+                if os.path.exists(log):
+                    assert os.path.getsize(log) >= 0
+                    shutil.copyfile(log, os.path.join(logdir, n + ".log"))
+                if os.path.exists(debuglog):
+                    assert os.path.getsize(debuglog) >= 0
+                    shutil.copyfile(debuglog, os.path.join(logdir, n + "_debug.log"))
+                if os.path.exists(gclog):
+                    assert os.path.getsize(gclog) >= 0
+                    shutil.copyfile(gclog, os.path.join(logdir, n + "_gc.log"))
+                if os.path.exists(compactionlog):
+                    assert os.path.getsize(compactionlog) >= 0
+                    shutil.copyfile(compactionlog, os.path.join(logdir, n + "_compaction.log"))
+            if os.path.exists(name):
+                os.unlink(name)
+            if not is_win():
+                os.symlink(basedir, name)
+
+    def cql_connection(self, node, keyspace=None, user=None,
+                       password=None, compression=True, protocol_version=None, port=None, ssl_opts=None, **kwargs):
+
+        return self._create_session(node, keyspace, user, password, compression,
+                                    protocol_version, port=port, ssl_opts=ssl_opts, **kwargs)
+
+    def exclusive_cql_connection(self, node, keyspace=None, user=None,
+                                 password=None, compression=True, protocol_version=None, port=None, ssl_opts=None,
+                                 **kwargs):
+
+        node_ip = get_ip_from_node(node)
+        wlrr = WhiteListRoundRobinPolicy([node_ip])
+
+        return self._create_session(node, keyspace, user, password, compression,
+                                    protocol_version, port=port, ssl_opts=ssl_opts, load_balancing_policy=wlrr,
+                                    **kwargs)
+
+    def _create_session(self, node, keyspace, user, password, compression, protocol_version,
+                        port=None, ssl_opts=None, execution_profiles=None, **kwargs):
+        node_ip = get_ip_from_node(node)
+        if not port:
+            port = get_port_from_node(node)
+
+        if protocol_version is None:
+            protocol_version = get_eager_protocol_version(node.cluster.version())
+
+        if user is not None:
+            auth_provider = get_auth_provider(user=user, password=password)
+        else:
+            auth_provider = None
+
+        profiles = {EXEC_PROFILE_DEFAULT: make_execution_profile(**kwargs)
+                    } if not execution_profiles else execution_profiles
+
+        cluster = PyCluster([node_ip],
+                            auth_provider=auth_provider,
+                            compression=compression,
+                            protocol_version=protocol_version,
+                            port=port,
+                            ssl_options=ssl_opts,
+                            connect_timeout=15,
+                            allow_beta_protocol_version=True,
+                            execution_profiles=profiles)
+        session = cluster.connect(wait_for_all_pools=True)
+
+        if keyspace is not None:
+            session.set_keyspace(keyspace)
+
+        self.connections.append(session)
+        return session
+
+    def patient_cql_connection(self, node, keyspace=None,
+                               user=None, password=None, timeout=30, compression=True,
+                               protocol_version=None, port=None, ssl_opts=None, **kwargs):
+        """
+        Returns a connection after it stops throwing NoHostAvailables due to not being ready.
+
+        If the timeout is exceeded, the exception is raised.
+        """
+        if is_win():
+            timeout *= 2
+
+        expected_log_lines = ('Control connection failed to connect, shutting down Cluster:',
+                              '[control connection] Error connecting to ')
+        with log_filter('cassandra.cluster', expected_log_lines):
+            session = retry_till_success(
+                self.cql_connection,
+                node,
+                keyspace=keyspace,
+                user=user,
+                password=password,
+                timeout=timeout,
+                compression=compression,
+                protocol_version=protocol_version,
+                port=port,
+                ssl_opts=ssl_opts,
+                bypassed_exception=NoHostAvailable,
+                **kwargs
+            )
+
+        return session
+
+    def patient_exclusive_cql_connection(self, node, keyspace=None,
+                                         user=None, password=None, timeout=30, compression=True,
+                                         protocol_version=None, port=None, ssl_opts=None, **kwargs):
+        """
+        Returns a connection after it stops throwing NoHostAvailables due to not being ready.
+
+        If the timeout is exceeded, the exception is raised.
+        """
+        if is_win():
+            timeout *= 2
+
+        return retry_till_success(
+            self.exclusive_cql_connection,
+            node,
+            keyspace=keyspace,
+            user=user,
+            password=password,
+            timeout=timeout,
+            compression=compression,
+            protocol_version=protocol_version,
+            port=port,
+            ssl_opts=ssl_opts,
+            bypassed_exception=NoHostAvailable,
+            **kwargs
+        )
+
+    def check_logs_for_errors(self):
+        for node in self.cluster.nodelist():
+            errors = list(self.__filter_errors(
+                ['\n'.join(msg) for msg in node.grep_log_for_errors()]))
+            if len(errors) is not 0:
+                for error in errors:
+                    print("Unexpected error in {node_name} log, error: \n{error}".format(node_name=node.name, error=error))
+                return True
+
+    def __filter_errors(self, errors):
+        """Filter errors, removing those that match self.ignore_log_patterns"""
+        if not hasattr(self, 'ignore_log_patterns'):
+            self.ignore_log_patterns = []
+        for e in errors:
+            for pattern in self.ignore_log_patterns:
+                if re.search(pattern, e):
+                    break
+            else:
+                yield e
+
+    def get_jfr_jvm_args(self):
+        """
+        @return The JVM arguments required for attaching flight recorder to a Java process.
+        """
+        return ["-XX:+UnlockCommercialFeatures", "-XX:+FlightRecorder"]
+
+    def start_jfr_recording(self, nodes):
+        """
+        Start Java flight recorder provided the cluster was started with the correct jvm arguments.
+        """
+        for node in nodes:
+            p = subprocess.Popen(['jcmd', str(node.pid), 'JFR.start'],
+                                 stdout=subprocess.PIPE,
+                                 stderr=subprocess.PIPE)
+            stdout, stderr = p.communicate()
+            logger.debug(stdout)
+            logger.debug(stderr)
+
+    def dump_jfr_recording(self, nodes):
+        """
+        Save Java flight recorder results to file for analyzing with mission control.
+        """
+        for node in nodes:
+            p = subprocess.Popen(['jcmd', str(node.pid), 'JFR.dump',
+                                  'recording=1', 'filename=recording_{}.jfr'.format(node.address())],
+                                 stdout=subprocess.PIPE,
+                                 stderr=subprocess.PIPE)
+            stdout, stderr = p.communicate()
+            logger.debug(stdout)
+            logger.debug(stderr)
+
+    def supports_v5_protocol(self, cluster_version):
+        return cluster_version >= LooseVersion('4.0')
+
+    def cleanup_last_test_dir(self):
+        if os.path.exists(self.last_test_dir):
+            os.remove(self.last_test_dir)
+
+    def stop_active_log_watch(self):
+        """
+        Joins the log watching thread, which will then exit.
+        Should be called after each test, ideally after nodes are stopped but before cluster files are removed.
+
+        Can be called multiple times without error.
+        If not called, log watching thread will remain running until the parent process exits.
+        """
+        self.log_watch_thread.join(timeout=60)
+
+    def cleanup_cluster(self):
+        with log_filter('cassandra'):  # quiet noise from driver when nodes start going down
+            if self.dtest_config.keep_test_dir:
+                self.cluster.stop(gently=self.dtest_config.enable_jacoco_code_coverage)
+            else:
+                # when recording coverage the jvm has to exit normally
+                # or the coverage information is not written by the jacoco agent
+                # otherwise we can just kill the process
+                if self.dtest_config.enable_jacoco_code_coverage:
+                    self.cluster.stop(gently=True)
+
+                # Cleanup everything:
+                try:
+                    if self.log_watch_thread:
+                        self.stop_active_log_watch()
+                finally:
+                    logger.debug("removing ccm cluster {name} at: {path}".format(name=self.cluster.name,
+                                                                          path=self.test_path))
+                    self.cluster.remove()
+
+                    logger.debug("clearing ssl stores from [{0}] directory".format(self.test_path))
+                    for filename in ('keystore.jks', 'truststore.jks', 'ccm_node.cer'):
+                        try:
+                            os.remove(os.path.join(self.test_path, filename))
+                        except OSError as e:
+                            # ENOENT = no such file or directory
+                            assert e.errno == errno.ENOENT
+
+                    os.rmdir(self.test_path)
+                    self.cleanup_last_test_dir()
+
+    def cleanup_and_replace_cluster(self):
+        for con in self.connections:
+            con.cluster.shutdown()
+        self.connections = []
+
+        self.cleanup_cluster()
+        self.test_path = self.get_test_path()
+        self.initialize_cluster()
+
+    def init_default_config(self):
+        # the failure detector can be quite slow in such tests with quick start/stop
+        phi_values = {'phi_convict_threshold': 5}
+
+        timeout = 15000
+        if self.cluster_options is not None and len(self.cluster_options) > 0:
+            values = merge_dicts(self.cluster_options, phi_values)
+        else:
+            values = merge_dicts(phi_values, {
+                'read_request_timeout_in_ms': timeout,
+                'range_request_timeout_in_ms': timeout,
+                'write_request_timeout_in_ms': timeout,
+                'truncate_request_timeout_in_ms': timeout,
+                'request_timeout_in_ms': timeout
+            })
+
+        if self.setup_overrides is not None and len(self.setup_overrides.cluster_options) > 0:
+            values = merge_dicts(values, self.setup_overrides.cluster_options)
+
+        # No more thrift in 4.0, and start_rpc doesn't exists anymore
+        if self.cluster.version() >= '4' and 'start_rpc' in values:
+            del values['start_rpc']
+
+        self.cluster.set_configuration_options(values)
+        logger.debug("Done setting configuration options:\n" + pprint.pformat(self.cluster._config_options, indent=4))
+
+    def maybe_setup_jacoco(self, cluster_name='test'):
+        """Setup JaCoCo code coverage support"""
+
+        if not self.dtest_config.enable_jacoco_code_coverage:
+            return
+
+        # use explicit agent and execfile locations
+        # or look for a cassandra build if they are not specified
+        agent_location = os.environ.get('JACOCO_AGENT_JAR',
+                                        os.path.join(self.dtest_config.cassandra_dir, 'build/lib/jars/jacocoagent.jar'))
+        jacoco_execfile = os.environ.get('JACOCO_EXECFILE',
+                                         os.path.join(self.dtest_config.cassandra_dir, 'build/jacoco/jacoco.exec'))
+
+        if os.path.isfile(agent_location):
+            logger.debug("Jacoco agent found at {}".format(agent_location))
+            with open(os.path.join(
+                    self.test_path, cluster_name, 'cassandra.in.sh'), 'w') as f:
+
+                f.write('JVM_OPTS="$JVM_OPTS -javaagent:{jar_path}=destfile={exec_file}"'
+                        .format(jar_path=agent_location, exec_file=jacoco_execfile))
+
+                if os.path.isfile(jacoco_execfile):
+                    logger.debug("Jacoco execfile found at {}, execution data will be appended".format(jacoco_execfile))
+                else:
+                    logger.debug("Jacoco execfile will be created at {}".format(jacoco_execfile))
+        else:
+            logger.debug("Jacoco agent not found or is not file. Execution will not be recorded.")
+
+    def create_ccm_cluster(self, name):
+        logger.debug("cluster ccm directory: " + self.test_path)
+        version = self.dtest_config.cassandra_version
+
+        if version:
+            cluster = Cluster(self.test_path, name, cassandra_version=version)
+        else:
+            cluster = Cluster(self.test_path, name, cassandra_dir=self.dtest_config.cassandra_dir)
+
+        if self.dtest_config.use_vnodes:
+            cluster.set_configuration_options(values={'initial_token': None, 'num_tokens': self.dtest_config.num_tokens})
+        else:
+            cluster.set_configuration_options(values={'num_tokens': None})
+
+        if self.dtest_config.use_off_heap_memtables:
+            cluster.set_configuration_options(values={'memtable_allocation_type': 'offheap_objects'})
+
+        cluster.set_datadir_count(self.dtest_config.data_dir_count)
+        cluster.set_environment_variable('CASSANDRA_LIBJEMALLOC', self.dtest_config.jemalloc_path)
+
+        return cluster
+
+    def set_cluster_log_levels(self):
+        """
+        The root logger gets configured in the fixture named fixture_logging_setup.
+        Based on the logging configuration options the user invoked pytest with,
+        that fixture sets the root logger to that configuration. We then ensure all
+        Cluster objects we work with "inherit" these logging settings (which we can
+        lookup off the root logger)
+        """
+        if logging.root.level != 'NOTSET':
+            log_level = logging.getLevelName(logging.INFO)
+        else:
+            log_level = logging.root.level
+        self.cluster.set_log_level(log_level)
+
+    def initialize_cluster(self):
+        """
+        This method is responsible for initializing and configuring a ccm
+        cluster for the next set of tests.  This can be called for two
+        different reasons:
+         * A class of tests is starting
+         * A test method failed/errored, so the cluster has been wiped
+
+        Subclasses that require custom initialization should generally
+        do so by overriding post_initialize_cluster().
+        """
+        # connections = []
+        # cluster_options = []
+        self.cluster = self.create_ccm_cluster(name='test')
+        self.init_default_config()
+        self.maybe_setup_jacoco()
+        self.set_cluster_log_levels()
+
+        # cls.init_config()
+        # write_last_test_file(cls.test_path, cls.cluster)
+
+        # cls.post_initialize_cluster()

http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/dtest_setup_overrides.py
----------------------------------------------------------------------
diff --git a/dtest_setup_overrides.py b/dtest_setup_overrides.py
new file mode 100644
index 0000000..6ea3258
--- /dev/null
+++ b/dtest_setup_overrides.py
@@ -0,0 +1,3 @@
+class DTestSetupOverrides:
+    def __init__(self):
+        self.cluster_options = []
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/global_row_key_cache_test.py
----------------------------------------------------------------------
diff --git a/global_row_key_cache_test.py b/global_row_key_cache_test.py
index c16f793..74c67f4 100644
--- a/global_row_key_cache_test.py
+++ b/global_row_key_cache_test.py
@@ -1,13 +1,16 @@
 import time
+import logging
 
 from cassandra.concurrent import execute_concurrent_with_args
 
-from dtest import Tester, debug, create_ks
+from dtest import Tester, create_ks, create_cf_simple
+
+logger = logging.getLogger(__name__)
 
 
 class TestGlobalRowKeyCache(Tester):
 
-    def functional_test(self):
+    def test_functional(self):
         cluster = self.cluster
         cluster.populate(3)
         node1 = cluster.nodelist()[0]
@@ -15,7 +18,7 @@ class TestGlobalRowKeyCache(Tester):
         for keycache_size in (0, 10):
             for rowcache_size in (0, 10):
                 cluster.stop()
-                debug("Testing with keycache size of %d MB, rowcache size of %d MB " %
+                logger.debug("Testing with keycache size of %d MB, rowcache size of %d MB " %
                       (keycache_size, rowcache_size))
                 keyspace_name = 'ks_%d_%d' % (keycache_size, rowcache_size)
 
@@ -29,14 +32,15 @@ class TestGlobalRowKeyCache(Tester):
 
                 cluster.start()
                 session = self.patient_cql_connection(node1)
-
                 create_ks(session, keyspace_name, rf=3)
 
                 session.set_keyspace(keyspace_name)
-                session.execute("CREATE TABLE test (k int PRIMARY KEY, v1 int, v2 int)")
-                session.execute("CREATE TABLE test_clustering (k int, v1 int, v2 int, PRIMARY KEY (k, v1))")
-                session.execute("CREATE TABLE test_counter (k int PRIMARY KEY, v1 counter)")
-                session.execute("CREATE TABLE test_counter_clustering (k int, v1 int, v2 counter, PRIMARY KEY (k, v1))")
+                create_cf_simple(session, 'test', "CREATE TABLE test (k int PRIMARY KEY, v1 int, v2 int)")
+                create_cf_simple(session, 'test_clustering',
+                                 "CREATE TABLE test_clustering (k int, v1 int, v2 int, PRIMARY KEY (k, v1))")
+                create_cf_simple(session, 'test_counter', "CREATE TABLE test_counter (k int PRIMARY KEY, v1 counter)")
+                create_cf_simple(session, 'test_counter_clustering',
+                                 "CREATE TABLE test_counter_clustering (k int, v1 int, v2 counter, PRIMARY KEY (k, v1))")
 
                 # insert 100 rows into each table
                 for cf in ('test', 'test_clustering'):
@@ -87,12 +91,12 @@ class TestGlobalRowKeyCache(Tester):
                 session.shutdown()
 
                 # let the data be written to the row/key caches.
-                debug("Letting caches be saved to disk")
+                logger.debug("Letting caches be saved to disk")
                 time.sleep(10)
-                debug("Stopping cluster")
+                logger.debug("Stopping cluster")
                 cluster.stop()
                 time.sleep(1)
-                debug("Starting cluster")
+                logger.debug("Starting cluster")
                 cluster.start()
                 time.sleep(5)  # read the data back from row and key caches
 
@@ -108,38 +112,38 @@ class TestGlobalRowKeyCache(Tester):
             rows = list(session.execute("SELECT * FROM %s" % (cf,)))
 
             # one row gets deleted each validation round
-            self.assertEquals(100 - (validation_round + 1), len(rows))
+            assert 100 - (validation_round + 1) == len(rows)
 
             # adjust enumeration start to account for row deletions
             for i, row in enumerate(sorted(rows), start=(validation_round + 1)):
-                self.assertEquals(i, row.k)
-                self.assertEquals(i, row.v1)
+                assert i == row.k
+                assert i == row.v1
 
                 # updated rows will have different values
                 expected_value = validation_round if i < num_updates else i
-                self.assertEquals(expected_value, row.v2)
+                assert expected_value == row.v2
 
         # check values of counter tables
         rows = list(session.execute("SELECT * FROM test_counter"))
-        self.assertEquals(100, len(rows))
+        assert 100 == len(rows)
         for i, row in enumerate(sorted(rows)):
-            self.assertEquals(i, row.k)
+            assert i == row.k
 
             # updated rows will get incremented once each round
             expected_value = i
             if i < num_updates:
                 expected_value += validation_round + 1
 
-            self.assertEquals(expected_value, row.v1)
+            assert expected_value == row.v1
 
         rows = list(session.execute("SELECT * FROM test_counter_clustering"))
-        self.assertEquals(100, len(rows))
+        assert 100 == len(rows)
         for i, row in enumerate(sorted(rows)):
-            self.assertEquals(i, row.k)
-            self.assertEquals(i, row.v1)
+            assert i == row.k
+            assert i == row.v1
 
             expected_value = i
             if i < num_updates:
                 expected_value += validation_round + 1
 
-            self.assertEquals(expected_value, row.v2)
+            assert expected_value == row.v2

http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/hintedhandoff_test.py
----------------------------------------------------------------------
diff --git a/hintedhandoff_test.py b/hintedhandoff_test.py
index 6345e3c..68d341e 100644
--- a/hintedhandoff_test.py
+++ b/hintedhandoff_test.py
@@ -1,11 +1,15 @@
 import os
 import time
+import pytest
+import logging
 
 from cassandra import ConsistencyLevel
 
-from dtest import DISABLE_VNODES, Tester, create_ks
+from dtest import Tester, create_ks
 from tools.data import create_c1c2_table, insert_c1c2, query_c1c2
-from tools.decorators import no_vnodes, since
+
+since = pytest.mark.since
+logger = logging.getLogger(__name__)
 
 
 @since('3.0')
@@ -26,7 +30,7 @@ class TestHintedHandoffConfig(Tester):
         if config_options:
             cluster.set_configuration_options(values=config_options)
 
-        if DISABLE_VNODES:
+        if not self.dtest_config.use_vnodes:
             cluster.populate([2]).start()
         else:
             tokens = cluster.balanced_tokens(2)
@@ -39,7 +43,7 @@ class TestHintedHandoffConfig(Tester):
         Launch a nodetool command and check there is no error, return the result
         """
         out, err, _ = node.nodetool(cmd)
-        self.assertEqual('', err)
+        assert '' == err
         return out
 
     def _do_hinted_handoff(self, node1, node2, enabled, keyspace='ks'):
@@ -65,13 +69,13 @@ class TestHintedHandoffConfig(Tester):
 
         # Check node2 for all the keys that should have been delivered via HH if enabled or not if not enabled
         session = self.patient_exclusive_cql_connection(node2, keyspace=keyspace)
-        for n in xrange(0, 100):
+        for n in range(0, 100):
             if enabled:
                 query_c1c2(session, n, ConsistencyLevel.ONE)
             else:
                 query_c1c2(session, n, ConsistencyLevel.ONE, tolerate_missing=True, must_be_missing=True)
 
-    def nodetool_test(self):
+    def test_nodetool(self):
         """
         Test various nodetool commands
         """
@@ -79,25 +83,25 @@ class TestHintedHandoffConfig(Tester):
 
         for node in node1, node2:
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running', res.rstrip())
+            assert 'Hinted handoff is running' == res.rstrip()
 
             self._launch_nodetool_cmd(node, 'disablehandoff')
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is not running', res.rstrip())
+            assert 'Hinted handoff is not running' == res.rstrip()
 
             self._launch_nodetool_cmd(node, 'enablehandoff')
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running', res.rstrip())
+            assert 'Hinted handoff is running' == res.rstrip()
 
             self._launch_nodetool_cmd(node, 'disablehintsfordc dc1')
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running{}Data center dc1 is disabled'.format(os.linesep), res.rstrip())
+            assert 'Hinted handoff is running{}Data center dc1 is disabled'.format(os.linesep) == res.rstrip()
 
             self._launch_nodetool_cmd(node, 'enablehintsfordc dc1')
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running', res.rstrip())
+            assert 'Hinted handoff is running' == res.rstrip()
 
-    def hintedhandoff_disabled_test(self):
+    def test_hintedhandoff_disabled(self):
         """
         Test gloabl hinted handoff disabled
         """
@@ -105,11 +109,11 @@ class TestHintedHandoffConfig(Tester):
 
         for node in node1, node2:
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is not running', res.rstrip())
+            assert 'Hinted handoff is not running' == res.rstrip()
 
         self._do_hinted_handoff(node1, node2, False)
 
-    def hintedhandoff_enabled_test(self):
+    def test_hintedhandoff_enabled(self):
         """
         Test global hinted handoff enabled
         """
@@ -117,12 +121,12 @@ class TestHintedHandoffConfig(Tester):
 
         for node in node1, node2:
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running', res.rstrip())
+            assert 'Hinted handoff is running' == res.rstrip()
 
         self._do_hinted_handoff(node1, node2, True)
 
     @since('4.0')
-    def hintedhandoff_setmaxwindow_test(self):
+    def test_hintedhandoff_setmaxwindow(self):
         """
         Test global hinted handoff against max_hint_window_in_ms update via nodetool
         """
@@ -130,18 +134,18 @@ class TestHintedHandoffConfig(Tester):
 
         for node in node1, node2:
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running', res.rstrip())
+            assert 'Hinted handoff is running' == res.rstrip()
 
         res = self._launch_nodetool_cmd(node, 'getmaxhintwindow')
-        self.assertEqual('Current max hint window: 300000 ms', res.rstrip())
+        assert 'Current max hint window: 300000 ms' == res.rstrip()
         self._do_hinted_handoff(node1, node2, True)
         node1.start(wait_other_notice=True)
         self._launch_nodetool_cmd(node, 'setmaxhintwindow 1')
         res = self._launch_nodetool_cmd(node, 'getmaxhintwindow')
-        self.assertEqual('Current max hint window: 1 ms', res.rstrip())
+        assert 'Current max hint window: 1 ms' == res.rstrip()
         self._do_hinted_handoff(node1, node2, False, keyspace='ks2')
 
-    def hintedhandoff_dc_disabled_test(self):
+    def test_hintedhandoff_dc_disabled(self):
         """
         Test global hinted handoff enabled with the dc disabled
         """
@@ -150,11 +154,11 @@ class TestHintedHandoffConfig(Tester):
 
         for node in node1, node2:
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running{}Data center dc1 is disabled'.format(os.linesep), res.rstrip())
+            assert 'Hinted handoff is running{}Data center dc1 is disabled'.format(os.linesep) == res.rstrip()
 
         self._do_hinted_handoff(node1, node2, False)
 
-    def hintedhandoff_dc_reenabled_test(self):
+    def test_hintedhandoff_dc_reenabled(self):
         """
         Test global hinted handoff enabled with the dc disabled first and then re-enabled
         """
@@ -163,20 +167,20 @@ class TestHintedHandoffConfig(Tester):
 
         for node in node1, node2:
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running{}Data center dc1 is disabled'.format(os.linesep), res.rstrip())
+            assert 'Hinted handoff is running{}Data center dc1 is disabled'.format(os.linesep) == res.rstrip()
 
         for node in node1, node2:
             self._launch_nodetool_cmd(node, 'enablehintsfordc dc1')
             res = self._launch_nodetool_cmd(node, 'statushandoff')
-            self.assertEqual('Hinted handoff is running', res.rstrip())
+            assert 'Hinted handoff is running' == res.rstrip()
 
         self._do_hinted_handoff(node1, node2, True)
 
 
 class TestHintedHandoff(Tester):
 
-    @no_vnodes()
-    def hintedhandoff_decom_test(self):
+    @pytest.mark.no_vnodes
+    def test_hintedhandoff_decom(self):
         self.cluster.populate(4).start(wait_for_binary_proto=True)
         [node1, node2, node3, node4] = self.cluster.nodelist()
         session = self.patient_cql_connection(node1)
@@ -192,5 +196,5 @@ class TestHintedHandoff(Tester):
         node3.decommission(force=force)
 
         time.sleep(5)
-        for x in xrange(0, 100):
+        for x in range(0, 100):
             query_c1c2(session, x, ConsistencyLevel.ONE)

http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/internode_ssl_test.py
----------------------------------------------------------------------
diff --git a/internode_ssl_test.py b/internode_ssl_test.py
index 4149d26..fff9985 100644
--- a/internode_ssl_test.py
+++ b/internode_ssl_test.py
@@ -1,11 +1,15 @@
-from dtest import Tester, debug, create_ks, create_cf
+import logging
+
+from dtest import Tester, create_ks, create_cf
 from tools.data import putget
 from tools.misc import generate_ssl_stores
 
+logger = logging.getLogger(__name__)
+
 
 class TestInternodeSSL(Tester):
 
-    def putget_with_internode_ssl_test(self):
+    def test_putget_with_internode_ssl(self):
         """
         Simple putget test with internode ssl enabled
         with default 'all' internode compression
@@ -13,7 +17,7 @@ class TestInternodeSSL(Tester):
         """
         self.__putget_with_internode_ssl_test('all')
 
-    def putget_with_internode_ssl_without_compression_test(self):
+    def test_putget_with_internode_ssl_without_compression(self):
         """
         Simple putget test with internode ssl enabled
         without internode compression
@@ -24,10 +28,10 @@ class TestInternodeSSL(Tester):
     def __putget_with_internode_ssl_test(self, internode_compression):
         cluster = self.cluster
 
-        debug("***using internode ssl***")
-        generate_ssl_stores(self.test_path)
+        logger.debug("***using internode ssl***")
+        generate_ssl_stores(self.fixture_dtest_setup.test_path)
         cluster.set_configuration_options({'internode_compression': internode_compression})
-        cluster.enable_internode_ssl(self.test_path)
+        cluster.enable_internode_ssl(self.fixture_dtest_setup.test_path)
 
         cluster.populate(3).start()
 

http://git-wip-us.apache.org/repos/asf/cassandra-dtest/blob/49b2dda4/jmx_auth_test.py
----------------------------------------------------------------------
diff --git a/jmx_auth_test.py b/jmx_auth_test.py
index 99e7d80..99b227f 100644
--- a/jmx_auth_test.py
+++ b/jmx_auth_test.py
@@ -1,16 +1,19 @@
+import pytest
+import logging
 from distutils.version import LooseVersion
 
 from ccmlib.node import ToolError
-
 from dtest import Tester
-from tools.decorators import since
 from tools.jmxutils import apply_jmx_authentication
 
+since = pytest.mark.since
+logger = logging.getLogger(__name__)
+
 
 @since('3.6')
 class TestJMXAuth(Tester):
 
-    def basic_auth_test(self):
+    def test_basic_auth(self):
         """
         Some basic smoke testing of JMX authentication and authorization.
         Uses nodetool as a means of exercising the JMX interface as JolokiaAgent
@@ -29,21 +32,21 @@ class TestJMXAuth(Tester):
         session.execute("GRANT DESCRIBE ON ALL MBEANS TO jmx_user")
         session.execute("CREATE ROLE test WITH LOGIN=true and PASSWORD='abc123'")
 
-        with self.assertRaisesRegexp(ToolError, self.authentication_fail_message(node, 'baduser')):
+        with pytest.raises(ToolError, matches=self.authentication_fail_message(node, 'baduser')):
             node.nodetool('-u baduser -pw abc123 gossipinfo')
 
-        with self.assertRaisesRegexp(ToolError, self.authentication_fail_message(node, 'test')):
+        with pytest.raises(ToolError, matches=self.authentication_fail_message(node, 'test')):
             node.nodetool('-u test -pw badpassword gossipinfo')
 
-        with self.assertRaisesRegexp(ToolError, "Required key 'username' is missing"):
+        with pytest.raises(ToolError, matches="Required key 'username' is missing"):
             node.nodetool('gossipinfo')
 
         # role must have LOGIN attribute
-        with self.assertRaisesRegexp(ToolError, 'jmx_user is not permitted to log in'):
+        with pytest.raises(ToolError, matches='jmx_user is not permitted to log in'):
             node.nodetool('-u jmx_user -pw 321cba gossipinfo')
 
         # test doesn't yet have any privileges on the necessary JMX resources
-        with self.assertRaisesRegexp(ToolError, 'Access Denied'):
+        with pytest.raises(ToolError, matches='Access Denied'):
             node.nodetool('-u test -pw abc123 gossipinfo')
 
         session.execute("GRANT jmx_user TO test")


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message