#!/usr/bin/env python

"""Common import for all tests."""

import base64
import json
import logging
import optparse
import os
import shlex
import shutil
import signal
import socket
import subprocess
import sys
import time
import unittest
import urllib2

import environment
from mysql_flavor import mysql_flavor
from mysql_flavor import set_mysql_flavor
import MySQLdb
from protocols_flavor import protocols_flavor
from topo_flavor.server import set_topo_server_flavor
from vtctl import vtctl_client
from vtdb import keyrange_constants
from vtgate_gateway_flavor.gateway import set_vtgate_gateway_flavor
from vtgate_gateway_flavor.gateway import vtgate_gateway_flavor
from vtproto import topodata_pb2


options = None
devnull = open('/dev/null', 'w')
hostname = socket.getaddrinfo(
    socket.getfqdn(), None, 0, 0, 0, socket.AI_CANONNAME)[0][3]


class TestError(Exception):
  pass


class Break(Exception):
  pass

environment.setup()


class LoggingStream(object):

  def __init__(self):
    self.line = ''

  def write(self, value):
    if value == '\n':
      # we already printed it
      self.line = ''
      return
    self.line += value
    logging.info('===== ' + self.line)
    if value.endswith('\n'):
      self.line = ''

  def writeln(self, value):
    self.write(value)
    self.line = ''

  def flush(self):
    pass


def add_options(parser):
  environment.add_options(parser)
  parser.add_option('-d', '--debug', action='store_true',
                    help='utils.pause() statements will wait for user input')
  parser.add_option('-k', '--keep-logs', action='store_true',
                    help='Do not delete log files on teardown.')
  parser.add_option(
      '-q', '--quiet', action='store_const', const=0, dest='verbose', default=1)
  parser.add_option(
      '-v', '--verbose', action='store_const', const=2, dest='verbose',
      default=1)
  parser.add_option('--skip-build', action='store_true',
                    help='Do not build the go binaries when running the test.')
  parser.add_option(
      '--skip-teardown', action='store_true',
      help='Leave the global processes running after the test is done.')
  parser.add_option('--mysql-flavor')
  parser.add_option('--protocols-flavor', default='grpc')
  parser.add_option('--topo-server-flavor', default='zookeeper')
  parser.add_option('--vtgate-gateway-flavor', default='discoverygateway')


def set_options(opts):
  global options
  options = opts

  set_mysql_flavor(options.mysql_flavor)
  environment.setup_protocol_flavor(options.protocols_flavor)
  set_topo_server_flavor(options.topo_server_flavor)
  set_vtgate_gateway_flavor(options.vtgate_gateway_flavor)
  environment.skip_build = options.skip_build


# main executes the test classes contained in the passed module, or
# __main__ if empty.
def main(mod=None, test_options=None):
  """The replacement main method, which parses args and runs tests.

  Args:
    mod: module that contains the test methods.
    test_options: a function which adds OptionParser options that are specific
      to a test file.
  """
  if mod is None:
    mod = sys.modules['__main__']

  global options

  parser = optparse.OptionParser(usage='usage: %prog [options] [test_names]')
  add_options(parser)
  if test_options:
    test_options(parser)
  (options, args) = parser.parse_args()

  set_log_level(options.verbose)
  logging.basicConfig(
      format='-- %(asctime)s %(module)s:%(lineno)d %(levelname)s %(message)s')

  set_options(options)

  run_tests(mod, args)


def run_tests(mod, args):
  try:
    suite = unittest.TestSuite()
    if not args:
      # this will run the setup and teardown
      suite.addTests(unittest.TestLoader().loadTestsFromModule(mod))
    else:
      if args[0] == 'teardown':
        mod.tearDownModule()

      elif args[0] == 'setup':
        mod.setUpModule()

      else:
        for arg in args:
          # this will run the setup and teardown
          suite.addTests(unittest.TestLoader().loadTestsFromName(arg, mod))

    if suite.countTestCases() > 0:
      logger = LoggingStream()
      result = unittest.TextTestRunner(
          stream=logger, verbosity=options.verbose, failfast=True).run(suite)
      if not result.wasSuccessful():
        sys.exit(-1)
  except KeyboardInterrupt:
    logging.warning('======== Tests interrupted, cleaning up ========')
    mod.tearDownModule()
    # If you interrupt a test, you probably want to stop evaluating the rest.
    sys.exit(1)
  finally:
    if options.keep_logs:
      logging.warning('Leaving temporary files behind (--keep-logs), please '
                      'clean up before next run: ' + os.environ['VTDATAROOT'])


def remove_tmp_files():
  if options.keep_logs:
    return
  try:
    shutil.rmtree(environment.tmproot)
  except OSError as e:
    logging.debug('remove_tmp_files: %s', str(e))


def pause(prompt):
  if options.debug:
    raw_input(prompt)


# sub-process management
pid_map = {}
already_killed = []


def _add_proc(proc):
  pid_map[proc.pid] = proc
  with open(environment.tmproot+'/test-pids', 'a') as f:
    print >> f, proc.pid, os.path.basename(proc.args[0])


def required_teardown():
  """Required cleanup steps that can't be skipped with --skip-teardown."""
  # We can't skip closing of gRPC connections, because the Python interpreter
  # won't let us die if any connections are left open.
  global vtctld_connection
  if vtctld_connection:
    vtctld_connection.close()
    vtctld_connection = None


def kill_sub_processes():
  for proc in pid_map.values():
    if proc.pid and proc.returncode is None:
      proc.kill()
  if not os.path.exists(environment.tmproot+'/test-pids'):
    return
  with open(environment.tmproot+'/test-pids') as f:
    for line in f:
      try:
        parts = line.strip().split()
        pid = int(parts[0])
        proc = pid_map.get(pid)
        if not proc or (proc and proc.pid and proc.returncode is None):
          if pid not in already_killed:
            os.kill(pid, signal.SIGTERM)
      except OSError as e:
        logging.debug('kill_sub_processes: %s', str(e))


def kill_sub_process(proc, soft=False):
  if proc is None:
    return
  pid = proc.pid
  if soft:
    proc.terminate()
  else:
    proc.kill()
  if pid and pid in pid_map:
    del pid_map[pid]
    already_killed.append(pid)


# run in foreground, possibly capturing output
def run(cmd, trap_output=False, raise_on_error=True, **kargs):
  if isinstance(cmd, str):
    args = shlex.split(cmd)
  else:
    args = cmd
  if trap_output:
    kargs['stdout'] = subprocess.PIPE
    kargs['stderr'] = subprocess.PIPE
  logging.debug(
      'run: %s %s', str(cmd),
      ', '.join('%s=%s' % x for x in kargs.iteritems()))
  proc = subprocess.Popen(args, **kargs)
  proc.args = args
  stdout, stderr = proc.communicate()
  if proc.returncode:
    if raise_on_error:
      pause('cmd fail: %s, pausing...' % (args))
      raise TestError('cmd fail:', args, proc.returncode, stdout, stderr)
    else:
      logging.debug('cmd fail: %s %d %s %s',
                    str(args), proc.returncode, stdout, stderr)
  return stdout, stderr


# run sub-process, expects failure
def run_fail(cmd, **kargs):
  if isinstance(cmd, str):
    args = shlex.split(cmd)
  else:
    args = cmd
  kargs['stdout'] = subprocess.PIPE
  kargs['stderr'] = subprocess.PIPE
  if options.verbose == 2:
    logging.debug(
        'run: (expect fail) %s %s', cmd,
        ', '.join('%s=%s' % x for x in kargs.iteritems()))
  proc = subprocess.Popen(args, **kargs)
  proc.args = args
  stdout, stderr = proc.communicate()
  if proc.returncode == 0:
    logging.info('stdout:\n%sstderr:\n%s', stdout, stderr)
    raise TestError('expected fail:', args, stdout, stderr)
  return stdout, stderr


# run a daemon - kill when this script exits
def run_bg(cmd, **kargs):
  if options.verbose == 2:
    logging.debug(
        'run: %s %s', cmd, ', '.join('%s=%s' % x for x in kargs.iteritems()))
  if 'extra_env' in kargs:
    kargs['env'] = os.environ.copy()
    if kargs['extra_env']:
      kargs['env'].update(kargs['extra_env'])
    del kargs['extra_env']
  if isinstance(cmd, str):
    args = shlex.split(cmd)
  else:
    args = cmd
  proc = subprocess.Popen(args=args, **kargs)
  proc.args = args
  _add_proc(proc)
  return proc


def wait_procs(proc_list, raise_on_error=True):
  for proc in proc_list:
    pid = proc.pid
    if pid:
      already_killed.append(pid)
  for proc in proc_list:
    proc.wait()
  for proc in proc_list:
    if proc.returncode:
      if options.verbose >= 1 and proc.returncode not in (-9,):
        sys.stderr.write('proc failed: %s %s\n' % (proc.returncode, proc.args))
      if raise_on_error:
        raise subprocess.CalledProcessError(proc.returncode,
                                            ' '.join(proc.args))


def validate_topology(ping_tablets=False):
  if ping_tablets:
    run_vtctl(['Validate', '-ping-tablets'])
  else:
    run_vtctl(['Validate'])


def zk_ls(path):
  out, _ = run(environment.binary_argstr('zk')+' ls '+path, trap_output=True)
  return sorted(out.splitlines())


def zk_cat(path):
  out, _ = run(environment.binary_argstr('zk')+' cat '+path, trap_output=True)
  return out


def zk_cat_json(path):
  data = zk_cat(path)
  return json.loads(data)


# wait_step is a helper for looping until a condition is true.
# use as follow:
#    timeout = 10
#    while True:
#      if done:
#        break
#      timeout = utils.wait_step('condition', timeout)
def wait_step(msg, timeout, sleep_time=0.1):
  timeout -= sleep_time
  if timeout <= 0:
    raise TestError('timeout waiting for condition "%s"' % msg)
  logging.debug('Sleeping for %f seconds waiting for condition "%s"',
                sleep_time, msg)
  time.sleep(sleep_time)
  return timeout


# vars helpers
def get_vars(port):
  """Returns the dict for vars from a vtxxx process. None if not available."""
  try:
    url = 'http://localhost:%d/debug/vars' % int(port)
    f = urllib2.urlopen(url)
    data = f.read()
    f.close()
  except urllib2.URLError:
    return None
  try:
    return json.loads(data)
  except ValueError:
    print data
    raise


def wait_for_vars(name, port, var=None, key=None, value=None, timeout=10.0):
  """Waits for the vars of a process, and optional values.

  Args:
    name: nickname for the process.
    port: process port to look at.
    var: if specified, waits for var in vars.
    key: if specified, waits for vars[var][key]==value.
    value: if key if specified, waits for vars[var][key]==value.
    timeout: how long to wait.
  """
  text = 'waiting for http://localhost:%d/debug/vars of %s' % (port, name)
  if var:
    text += ' value %s' % var
    if key:
      text += ' key %s:%s' % (key, value)
  while True:
    display_text = text
    v = get_vars(port)
    if v:
      if var is None:
        break
      if var in v:
        if key is None:
          break
        if key in v[var]:
          if v[var][key] == value:
            break
          else:
            display_text += ' (current value:%s)' % v[var][key]
        else:
          display_text += ' (no current value)'
      else:
        display_text += ' (%s not in vars)' % var
    else:
      display_text += ' (no vars yet)'
    timeout = wait_step(display_text, timeout)


def poll_for_vars(
    name, port, condition_msg, timeout=60.0, condition_fn=None,
    require_vars=False):
  """Polls for debug variables to exist or match specific conditions.

  This function polls in a tight loop, with no sleeps. This is useful for
  variables that are expected to be short-lived (e.g., a 'Done' state
  immediately before a process exits).

  Args:
    name: the name of the process that we're trying to poll vars from.
    port: the port number that we should poll for variables.
    condition_msg: string describing the conditions that we're polling for,
      used for error messaging.
    timeout: number of seconds that we should attempt to poll for.
    condition_fn: a function that takes the debug vars dict as input, and
      returns a truthy value if it matches the success conditions.
    require_vars: True iff we expect the vars to always exist. If
      True, and the vars don't exist, we'll raise a TestError. This
      can be used to differentiate between a timeout waiting for a
      particular condition vs if the process that you're polling has
      already exited.

  Raises:
    TestError: if the conditions aren't met within the given timeout, or
               if vars are required and don't exist.

  Returns:
    dict of debug variables

  """
  start_time = time.time()
  while True:
    if (time.time() - start_time) >= timeout:
      raise TestError(
          'Timed out polling for vars from %s; condition "%s" not met' %
          (name, condition_msg))
    v = get_vars(port)
    if v is None:
      if require_vars:
        raise TestError(
            'Expected vars to exist on %s, but they do not; '
            'process probably exited earlier than expected.' % (name,))
      continue
    if condition_fn is None:
      return v
    elif condition_fn(v):
      return v


def apply_vschema(vschema):
  for k, v in vschema.iteritems():
    fname = os.path.join(environment.tmproot, 'vschema.json')
    with open(fname, 'w') as f:
      f.write(v)
    run_vtctl(['ApplyVSchema', '-vschema_file', fname, k])


def wait_for_tablet_type(tablet_alias, expected_type, timeout=10):
  """Waits for a given tablet's SlaveType to become the expected value.

  Args:
    tablet_alias: Alias of the tablet.
    expected_type: Type of the tablet e.g. "replica".
    timeout: Timeout in seconds.

  Raises:
    TestError: SlaveType did not become expected_type within timeout seconds.
  """
  type_as_int = topodata_pb2.TabletType.Value(expected_type.upper())
  while True:
    if run_vtctl_json(['GetTablet', tablet_alias])['type'] == type_as_int:
      logging.debug('tablet %s went to expected type: %s',
                    tablet_alias, expected_type)
      break
    timeout = wait_step(
        "%s's SlaveType to be %s" % (tablet_alias, expected_type),
        timeout)


def wait_for_replication_pos(tablet_a, tablet_b, timeout=60.0):
  """Waits for tablet B to catch up to the replication position of tablet A.

  Args:
    tablet_a: tablet Object for tablet A.
    tablet_b: tablet Object for tablet B.
    timeout: Timeout in seconds.

  Raises:
    TestError: replication position did not catch up within timeout seconds.
  """
  replication_pos_a = mysql_flavor().master_position(tablet_a)
  while True:
    replication_pos_b = mysql_flavor().master_position(tablet_b)
    if mysql_flavor().position_at_least(replication_pos_b, replication_pos_a):
      break
    timeout = wait_step(
        "%s's replication position to catch up %s's; "
        'currently at: %s, waiting to catch up to: %s' % (
            tablet_b.tablet_alias, tablet_a.tablet_alias, replication_pos_b,
            replication_pos_a),
        timeout, sleep_time=0.1)

# Save the first running instance of vtgate. It is saved when 'start'
# is called, and cleared when kill is called.
vtgate = None


class VtGate(object):
  """VtGate object represents a vtgate process."""

  def __init__(self, port=None):
    """Creates the Vtgate instance and reserve the ports if necessary."""
    self.port = port or environment.reserve_ports(1)
    if protocols_flavor().vtgate_protocol() == 'grpc':
      self.grpc_port = environment.reserve_ports(1)
    self.proc = None

  def start(self, cell='test_nj', retry_count=2,
            topo_impl=None, cache_ttl='1s',
            healthcheck_conn_timeout='2s',
            extra_args=None, tablets=None,
            tablet_types_to_wait='MASTER,REPLICA'):
    """Start vtgate. Saves it into the global vtgate variable if not set yet."""

    args = environment.binary_args('vtgate') + [
        '-port', str(self.port),
        '-cell', cell,
        '-retry-count', str(retry_count),
        '-log_dir', environment.vtlogroot,
        '-srv_topo_cache_ttl', cache_ttl,
        '-healthcheck_conn_timeout', healthcheck_conn_timeout,
        '-tablet_protocol', protocols_flavor().tabletconn_protocol(),
        '-gateway_implementation', vtgate_gateway_flavor().flavor(),
        '-tablet_grpc_combine_begin_execute',
    ]
    args.extend(vtgate_gateway_flavor().flags(cell=cell, tablets=tablets))
    if tablet_types_to_wait:
      args.extend(['-tablet_types_to_wait', tablet_types_to_wait])
    if protocols_flavor().vtgate_protocol() == 'grpc':
      args.extend(['-grpc_port', str(self.grpc_port)])
    if protocols_flavor().service_map():
      args.extend(['-service_map', ','.join(protocols_flavor().service_map())])
    if topo_impl:
      args.extend(['-topo_implementation', topo_impl])
    else:
      args.extend(environment.topo_server().flags())
    if extra_args:
      args.extend(extra_args)

    self.proc = run_bg(args)
    wait_for_vars('vtgate', self.port)

    global vtgate
    if not vtgate:
      vtgate = self

  def kill(self):
    """Terminates the vtgate process, and waits for it to exit.

    If this process is the one saved in the global vtgate variable,
    clears it.

    Note if the test is using just one global vtgate process, and
    starting it with the test, and killing it at the end of the test,
    there is no need to call this kill() method,
    utils.kill_sub_processes() will do a good enough job.

    """
    if self.proc is None:
      return
    kill_sub_process(self.proc, soft=True)
    self.proc.wait()
    self.proc = None

    global vtgate
    if vtgate == self:
      vtgate = None

  def addr(self):
    """Returns the address of the vtgate process, for web access."""
    return 'localhost:%d' % self.port

  def rpc_endpoint(self, python=False):
    """Returns the protocol and endpoint to use for RPCs."""
    if python:
      protocol = protocols_flavor().vtgate_python_protocol()
    else:
      protocol = protocols_flavor().vtgate_protocol()
    if protocol == 'grpc':
      return protocol, 'localhost:%d' % self.grpc_port
    return protocol, self.addr()

  def get_status(self):
    """Returns the status page for this process."""
    return get_status(self.port)

  def get_vars(self):
    """Returns the vars for this process."""
    return get_vars(self.port)

  def vtclient(self, sql, keyspace=None, shard=None, tablet_type='master',
               bindvars=None, streaming=False,
               verbose=False, raise_on_error=True, json_output=False):
    """Uses the vtclient binary to send a query to vtgate."""
    protocol, addr = self.rpc_endpoint()
    args = environment.binary_args('vtclient') + [
        '-server', addr,
        '-tablet_type', tablet_type,
        '-vtgate_protocol', protocol]
    if json_output:
      args.append('-json')
    if keyspace:
      args.extend(['-keyspace', keyspace])
    if shard:
      args.extend(['-shard', shard])