import logging
import os
import os.path
import pytest
import shutil
import string
import time

from ccmlib.node import TimeoutError
from distutils.version import LooseVersion
from dtest import Tester
from tools import sslkeygen

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

# see https://issues.apache.org/jira/browse/CASSANDRA-16127
class TestClientNetworkStopStart(Tester):

    def _normalize(self, a):
        return a.translate(str.maketrans(dict.fromkeys(string.whitespace)))

    def _in(self, a, b):
        return self._normalize(a) in self._normalize(b)

    def _assert_client_active_msg(self, name, enabled, out):
        expected = "{} active: {}".format(name, str(enabled).lower())
        actived = "actived" if enabled else "deactivated"
        assert self._in(expected, out), "{} is expected to be {} ({}) but was not found in output: {}".format(name, actived, str(enabled).lower(), out)

    def _assert_watch_log_for(self, node_or_cluster, to_watch, assert_msg=None):
        if assert_msg is None:
            assert_msg = "Unable to locate '{}'".format(to_watch)
        nodelist_fn = getattr(node_or_cluster, "nodelist", None)
        logger.debug("watching for '{}'".format(to_watch))
        start = time.perf_counter()
        if callable(nodelist_fn):
            for node in nodelist_fn():
                assert node.watch_log_for_no_errors(to_watch), assert_msg
        else:
            assert node_or_cluster.watch_log_for_no_errors(to_watch), assert_msg
        logger.debug("Completed watching for '{}'; took {}s".format(to_watch, time.perf_counter() - start))

    def _assert_binary_actually_found(self, node_or_cluster):
        # ccm will silently move on if the logs don't have CQL in time, which then leads to
        # flaky tests; to avoid that force waiting to be correct and assert the log was seen.
        logger.debug("Verifying that the CQL log was seen and that ccm didn't return early...")
        self._assert_watch_log_for(node_or_cluster, "Starting listening for CQL clients on", "Binary didn't start...")

    def _assert_client_enable(self, node, native_enabled=True, thrift_enabled=False):
        out = node.nodetool("info")
        self._assert_client_active_msg("Native Transport", native_enabled, out.stdout)
        if node.get_cassandra_version() >= LooseVersion('4.0'):
            assert "Thrift" not in out.stdout, "Thrift found in output: {}".format(out.stdout)
        else:
            self._assert_client_active_msg("Thrift", thrift_enabled, out.stdout)

    def _assert_startup(self, node_or_cluster):
        """Checks to see if the startup message was found"""
        self._assert_watch_log_for(node_or_cluster, "Startup complete", "Unable to find startup message, either the process crashed or is missing CASSANDRA-16127")

    @since(['2.2', '3.0.23', '3.11.9'])
    def test_defaults(self):
        """Tests default configurations have the correct client network setup"""
        cluster = self.cluster
        logger.debug("Starting cluster..")
        cluster.populate(1).start(wait_for_binary_proto=True)
        self._assert_binary_actually_found(cluster)
        self._assert_startup(cluster)
        node = cluster.nodelist()[0]
        self._assert_client_enable(node)

    @since(['2.2', '3.0.23', '3.11.9'], max_version='3.11.x')
    def test_hsha_defaults(self):
        """Enables hsha"""
        cluster = self.cluster
        logger.debug("Starting cluster..")
        cluster.set_configuration_options(values={
            'rpc_server_type': 'hsha',
            # seems 1 causes a dead lock... heh...
            'rpc_min_threads': 16,
            'rpc_max_threads': 2048
        })
        cluster.populate(1).start(wait_for_binary_proto=True)
        self._assert_binary_actually_found(cluster)
        self._assert_startup(cluster)
        node = cluster.nodelist()[0]
        self._assert_client_enable(node)

    @since(['2.2', '3.0.23', '3.11.9'], max_version='3.11.x')
    def test_hsha_with_ssl(self):
        """Enables hsha with ssl"""
        cluster = self.cluster
        logger.debug("Starting cluster..")
        cred = sslkeygen.generate_credentials("127.0.0.1")
        cluster.set_configuration_options(values={
            'rpc_server_type': 'hsha',
            # seems 1 causes a dead lock... heh...
            'rpc_min_threads': 16,
            'rpc_max_threads': 2048,
            'client_encryption_options': {
                'enabled': True,
                'optional': False,
                'keystore': cred.cakeystore,
                'keystore_password': 'cassandra'
            }
        })
        cluster.populate(1).start(wait_for_binary_proto=True)
        self._assert_binary_actually_found(cluster)
        self._assert_startup(cluster)
        node = cluster.nodelist()[0]
        self._assert_client_enable(node)