import glob
import os
import subprocess
import time
import pytest
import logging

from ccmlib import common
from ccmlib.node import ToolError

from dtest import Tester
from tools.intervention import InterruptCompaction

since = pytest.mark.since
logger = logging.getLogger(__name__)

# These must match the stress schema names
KeyspaceName = 'keyspace1'
TableName = 'standard1'


def _normcase_all(xs):
    """
    Return a list of the elements in xs, each with its casing normalized for
    use as a filename.
    """
    return [os.path.normcase(x) for x in xs]


@since('3.0')
class TestSSTableUtil(Tester):

    def test_compaction(self):
        """
        @jira_ticket CASSANDRA-7066

        Check that we can list sstable files after a successful compaction (no temporary sstable files)
        """
        cluster = self.cluster
        cluster.populate(1).start()
        node = cluster.nodelist()[0]

        self._create_data(node, KeyspaceName, TableName, 100000)
        finalfiles, tmpfiles = self._check_files(node, KeyspaceName, TableName)
        assert 0 == len(tmpfiles)

        node.compact()
        time.sleep(5)
        finalfiles, tmpfiles = self._check_files(node, KeyspaceName, TableName)
        assert 0 == len(tmpfiles)

    def test_abortedcompaction(self):
        """
        @jira_ticket CASSANDRA-7066
        @jira_ticket CASSANDRA-11497

        Check that we can cleanup temporary files after a compaction is aborted.
        """
        log_file_name = 'debug.log'
        cluster = self.cluster
        cluster.populate(1).start()
        node = cluster.nodelist()[0]

        numrecords = 250000

        self._create_data(node, KeyspaceName, TableName, numrecords)
        finalfiles, tmpfiles = self._check_files(node, KeyspaceName, TableName)
        assert len(finalfiles) > 0, "Expected to find some final files"
        assert 0 == len(tmpfiles), "Expected no tmp files"

        t = InterruptCompaction(node, TableName, filename=log_file_name, delay=2)
        t.start()

        try:
            logger.debug("Compacting...")
            node.compact()
        except ToolError:
            pass  # expected to fail

        t.join()

        finalfiles = _normcase_all(self._invoke_sstableutil(KeyspaceName, TableName, type='final'))
        tmpfiles = _normcase_all(self._invoke_sstableutil(KeyspaceName, TableName, type='tmp'))

        # In most cases we should end up with some temporary files to clean up, but it may happen
        # that no temporary files are created if compaction finishes too early or starts too late
        # see CASSANDRA-11497
        logger.debug("Got {} final files and {} tmp files after compaction was interrupted"
              .format(len(finalfiles), len(tmpfiles)))

        self._invoke_sstableutil(KeyspaceName, TableName, cleanup=True)

        self._check_files(node, KeyspaceName, TableName, finalfiles, [])

        # restart to make sure not data is lost
        logger.debug("Restarting node...")
        node.start(wait_for_binary_proto=True)
        # in some environments, a compaction may start that would change sstable files. We should wait if so
        node.wait_for_compactions()

        finalfiles, tmpfiles = self._check_files(node, KeyspaceName, TableName)
        assert 0 == len(tmpfiles)

        logger.debug("Running stress to ensure data is readable")
        self._read_data(node, numrecords)

    def _create_data(self, node, ks, table, numrecords):
        """
         This is just to create the schema so we can disable compaction
        """
        node.stress(['write', 'n=1', 'no-warmup', '-rate', 'threads=1'])
        node.nodetool('disableautocompaction %s %s' % (ks, table))

        node.stress(['write', 'n=%d' % (numrecords), 'no-warmup', '-rate', 'threads=50'])
        node.flush()

    def _read_data(self, node, numrecords):
        node.stress(['read', 'n=%d' % (numrecords,), 'no-warmup', '-rate', 'threads=25'])

    def _check_files(self, node, ks, table, expected_finalfiles=None, expected_tmpfiles=None):
        sstablefiles = _normcase_all(self._get_sstable_files(node, ks, table))
        allfiles = _normcase_all(self._invoke_sstableutil(ks, table, type='all'))
        finalfiles = _normcase_all(self._invoke_sstableutil(ks, table, type='final'))
        tmpfiles = _normcase_all(self._invoke_sstableutil(ks, table, type='tmp'))
        expected_oplogs = _normcase_all(self._get_sstable_transaction_logs(node, ks, table))
        tmpfiles_with_oplogs = _normcase_all(self._invoke_sstableutil(ks, table, type='tmp', oplogs=True))
        oplogs = _normcase_all(sorted(list(set(tmpfiles_with_oplogs) - set(tmpfiles))))

        if expected_finalfiles is None:
            expected_finalfiles = allfiles
        else:
            expected_finalfiles = _normcase_all(expected_finalfiles)

        if expected_tmpfiles is None:
            expected_tmpfiles = sorted(set(allfiles) - set(finalfiles))
        else:
            expected_tmpfiles = _normcase_all(expected_tmpfiles)

        logger.debug("Comparing all files...")
        assert sstablefiles == allfiles

        logger.debug("Comparing final files...")
        assert expected_finalfiles == finalfiles

        logger.debug("Comparing tmp files...")
        assert expected_tmpfiles == tmpfiles

        logger.debug("Comparing op logs...")
        assert expected_oplogs == oplogs

        return finalfiles, tmpfiles

    def _invoke_sstableutil(self, ks, table, type='all', oplogs=False, cleanup=False):
        """
        Invoke sstableutil and return the list of files, if any
        """
        logger.debug("About to invoke sstableutil with type {}...".format(type))
        node1 = self.cluster.nodelist()[0]
        env = common.make_cassandra_env(node1.get_install_cassandra_root(), node1.get_node_cassandra_root())
        tool_bin = node1.get_tool('sstableutil')

        args = [tool_bin, '--type', type]

        if oplogs:
            args.extend(['--oplog'])
        if cleanup:
            args.extend(['--cleanup'])

        args.extend([ks, table])

        p = subprocess.Popen(args, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        (stdout, stderr) = p.communicate()

        assert p.returncode == 0, "Error invoking sstableutil; returned {code}".format(code=p.returncode)

        if stdout:
            logger.debug(stdout.decode("utf-8"))

        match = ks + os.sep + table + '-'
        ret = sorted([s for s in stdout.decode("utf-8").splitlines() if match in s])
        logger.debug("Got {} files of type {}".format(len(ret), type))
        return ret

    def _get_sstable_files(self, node, ks, table):
        """
        Read sstable files directly from disk
        """
        ret = []
        for data_dir in node.data_directories():
            # note, the /var/folders -> /private/var/folders stuff is to fixup mac compatibility
            keyspace_dir = os.path.abspath(os.path.join(data_dir, ks)).replace("/var/folders", "/private/var/folders")
            for ext in ('*.db', '*.txt', '*.adler32', '*.crc32'):
                ret.extend(glob.glob(os.path.abspath(os.path.join(keyspace_dir, table + '-*', ext))))

        return sorted(ret)

    def _get_sstable_transaction_logs(self, node, ks, table):
        ret = []
        for data_dir in node.data_directories():
            keyspace_dir = os.path.join(data_dir, ks)
            ret.extend(glob.glob(os.path.join(keyspace_dir, table + '-*', "*.log")))

        return sorted(ret)